This commit is contained in:
Mateusz Gruszczyński
2026-09-14 16:32:28 +02:00
parent c3fbc5ccc6
commit 021cbddba5
68 changed files with 7780 additions and 202 deletions
+38 -1
View File
@@ -4,7 +4,9 @@ use crate::{
home_assistant, influxdb,
models::{
ApiTokenInfo, ApplicationSettings, Automation, ClimateGroup, ConfigurationExport,
DebugSettings, Device, DeviceCommand, DevicePatch, DiscoveryRequest, Flow, GreeSettings,
ConnectionStatus, ConnectionType, DebugSettings, Device, DeviceCommand, DevicePatch, DiscoveryRequest,
EnergySourcePreference, Flow, GreeSettings,
GreeCloudSettings, GreeCloudSettingsUpdate, GreeCloudSettingsView,
GroupControlPatch, HaReading, HistorySettings, HomeAssistantSettings,
HomeAssistantSettingsUpdate, HomeAssistantSettingsView, InfluxDbSettings,
InfluxDbSettingsUpdate, InfluxDbSettingsView, ManualDeviceRequest, NightModeSettings,
@@ -74,6 +76,7 @@ const SPA_ROUTES: &[&str] = &[
"/history/overview",
"/history/zones",
"/history/devices",
"/history/energy",
"/history/sensors",
"/history/custom",
];
@@ -147,6 +150,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/control-plan", get(control_plan))
.route("/api/events", get(events))
.route(
@@ -157,6 +161,10 @@ pub fn router(state: AppState) -> Router {
"/api/settings/gree",
get(get_gree_settings).put(update_gree_settings),
)
.route(
"/api/settings/gree-cloud",
get(get_gree_cloud_settings).put(update_gree_cloud_settings),
)
.route(
"/api/settings/history",
get(get_history_settings).put(update_history_settings),
@@ -191,6 +199,30 @@ pub fn router(state: AppState) -> Router {
"/api/access-tokens/:id",
axum::routing::delete(delete_access_token),
)
.route(
"/api/integrations/gree-cloud/test",
post(test_gree_cloud),
)
.route(
"/api/integrations/gree-cloud/devices",
get(discover_gree_cloud_devices),
)
.route(
"/api/integrations/gree-cloud/devices/:cloud_id/add",
post(add_gree_cloud_device),
)
.route(
"/api/integrations/gree-cloud/status",
get(gree_cloud_status),
)
.route(
"/api/integrations/gree-cloud/reconnect",
post(reconnect_gree_cloud),
)
.route(
"/api/devices/:id/cloud-diagnostics",
get(cloud_device_diagnostics),
)
.route(
"/api/integrations/home-assistant/test",
post(test_home_assistant),
@@ -199,6 +231,10 @@ pub fn router(state: AppState) -> Router {
"/api/integrations/home-assistant/entity",
post(inspect_home_assistant_entity),
)
.route(
"/api/integrations/home-assistant/energy-sensors",
get(list_home_assistant_energy_sensors),
)
.route(
"/api/integrations/notifications/test",
post(test_notifications),
@@ -320,6 +356,7 @@ include!("api/settings.rs");
include!("api/configuration.rs");
include!("api/debug_tokens.rs");
include!("api/integrations.rs");
include!("api/gree_cloud.rs");
include!("api/middleware.rs");
include!("api/websocket.rs");
include!("api/assets.rs");
+102 -5
View File
@@ -17,12 +17,16 @@ async fn export_configuration(
let settings = state.settings.read().await.clone();
let mut export = state.db.export_configuration(settings)?;
sanitize_configuration_runtime(&mut export);
// Cloud credentials are account-scoped secrets and must never be returned to the frontend,
// including configuration exports. Import can reuse the already configured password on the
// target controller when account/region match.
export.settings.gree_cloud.password.clear();
Ok(Json(export))
}
fn validate_configuration_header(export: &ConfigurationExport) -> Result<(), AppError> {
if export.format_version != 3 {
return Err(AppError::BadRequest("unsupported configuration export version; version 3 is required by GREE Controller 0.12.x".into()));
return Err(AppError::BadRequest("unsupported configuration export version; version 3 is required by GREE Controller 0.14.0".into()));
}
if export.settings.control_strategy != "setpoint" {
return Err(AppError::BadRequest(
@@ -108,14 +112,16 @@ fn validate_configuration_devices_and_zones(
export: &ConfigurationExport,
ids: &ConfigurationIds<'_>,
) -> Result<(), AppError> {
let device_macs: std::collections::HashSet<&str> = export
// The same physical unit may intentionally exist once as Local and once as GREE Cloud.
// Reject duplicates only within the same explicit transport.
let device_transport_macs: std::collections::HashSet<(ConnectionType, &str)> = export
.devices
.iter()
.map(|item| item.mac.as_str())
.map(|item| (item.connection_type, item.mac.as_str()))
.collect();
if device_macs.len() != export.devices.len() {
if device_transport_macs.len() != export.devices.len() {
return Err(AppError::BadRequest(
"import contains duplicate device MAC addresses".into(),
"import contains duplicate device MAC addresses for the same connection type".into(),
));
}
if export
@@ -557,6 +563,15 @@ fn normalize_imported_runtime_settings(settings: &mut RuntimeSettings) -> Result
settings.compressor_protection_enabled = gree.compressor_protection_enabled;
settings.compressor_protection_seconds = gree.compressor_protection_seconds;
if crate::protocol::gree_cloud::region_base_url(settings.gree_cloud.region.trim()).is_none() {
return Err(AppError::BadRequest("unsupported GREE Cloud region".into()));
}
settings.gree_cloud.polling_interval_seconds =
settings.gree_cloud.polling_interval_seconds.clamp(30, 3600);
if settings.gree_cloud.account_id.trim().is_empty() {
settings.gree_cloud.account_id = "default".into();
}
settings.history_retention_days = settings.history_retention_days.clamp(1, 3650);
settings.event_log_retention_days = settings.event_log_retention_days.clamp(1, 3650);
settings.influxdb.history_threshold_days =
@@ -601,6 +616,81 @@ fn prepare_configuration_import(export: &mut ConfigurationExport) -> Result<(),
Ok(())
}
async fn hydrate_imported_cloud_device_keys(
state: &AppState,
export: &mut ConfigurationExport,
) -> Result<(), AppError> {
if export.settings.gree_cloud.password.trim().is_empty() {
let current = state.settings.read().await.gree_cloud.clone();
if current.region.eq_ignore_ascii_case(&export.settings.gree_cloud.region)
&& current.username.eq_ignore_ascii_case(&export.settings.gree_cloud.username)
&& !current.password.trim().is_empty()
{
export.settings.gree_cloud.password = current.password;
}
}
let needs_cloud_keys = export.devices.iter().any(|device| {
device.connection_type == ConnectionType::GreeCloud
&& device.key.as_deref().unwrap_or_default().is_empty()
});
if !needs_cloud_keys {
return Ok(());
}
let settings = &export.settings.gree_cloud;
let mut api = crate::protocol::gree_cloud::GreeCloudApi::for_region(
state.http.clone(),
&settings.region,
&settings.username,
&settings.password,
)
.map_err(|err| AppError::BadRequest(format!(
"cannot restore GREE Cloud device secrets from account: {err}"
)))?;
api.login().await.map_err(|err| {
AppError::BadRequest(format!(
"cannot restore GREE Cloud device secrets: account login failed: {err}"
))
})?;
let discovered = api.get_all_devices().await.map_err(|err| {
AppError::BadRequest(format!(
"cannot restore GREE Cloud device secrets: discovery failed: {err}"
))
})?;
for device in export
.devices
.iter_mut()
.filter(|device| device.connection_type == ConnectionType::GreeCloud)
{
if device.key.as_deref().is_some_and(|key| !key.is_empty()) {
continue;
}
let cloud_id = device.cloud_device_id.as_deref().unwrap_or(&device.mac);
let Some(found) = discovered
.iter()
.find(|candidate| candidate.mac.eq_ignore_ascii_case(cloud_id))
else {
return Err(AppError::BadRequest(format!(
"cannot restore GREE Cloud device {}: it is not present in the configured account",
device.name
)));
};
device.key = Some(found.key.clone());
let normalized_cloud_mac = found.mac.replace([':', '-'], "").to_ascii_uppercase();
device.cloud_device_id = Some(normalized_cloud_mac.clone());
device.cloud_parent_mac = Some(crate::protocol::gree_cloud::parent_mac(&normalized_cloud_mac));
device.cloud_account_id = Some(settings.account_id.clone());
if device.model.trim().is_empty() {
device.model = found.model.clone().unwrap_or_default();
}
if device.firmware.trim().is_empty() {
device.firmware = found.version.clone().unwrap_or_default();
}
}
Ok(())
}
async fn lock_configuration_resources(
state: &AppState,
current_zones: &[Zone],
@@ -720,6 +810,7 @@ async fn import_configuration(
) -> Result<Json<Value>, AppError> {
validate_configuration_export(&export)?;
prepare_configuration_import(&mut export)?;
hydrate_imported_cloud_device_keys(&state, &mut export).await?;
let _configuration_guard = state.lock_configuration_operation().await;
let _automation_guard = state.lock_automation_operation().await;
@@ -740,6 +831,12 @@ async fn import_configuration(
state
.debug_gree_frames
.store(export.settings.debug.gree_frames, Ordering::Relaxed);
state
.debug_cloud_requests
.store(export.settings.debug.cloud_requests, Ordering::Relaxed);
state
.debug_cloud_mqtt
.store(export.settings.debug.cloud_mqtt, Ordering::Relaxed);
*state.settings.write().await = export.settings.clone();
reconcile_imported_devices(&state, &export).await?;
state
+125 -14
View File
@@ -12,7 +12,9 @@ async fn discover(
let protocol_version = request.protocol_version.unwrap_or(0).min(2);
let passes = request.passes.unwrap_or(3).clamp(1, 10);
let discovered = state
.gree
.providers
.local()
.client()
.discover(
&broadcast,
Duration::from_millis(timeout_ms),
@@ -38,7 +40,7 @@ async fn discover(
// Bind right after discovery. GREE modules can have a short bind window;
// bind() also refreshes it with a direct scan before the handshake.
if !merged.simulated && merged.key.as_deref().unwrap_or_default().is_empty() {
match state.gree.bind(&merged).await {
match state.providers.local().client().bind(&merged).await {
Ok(bound) => {
merged.key = Some(bound.key);
merged.protocol_version = bound.protocol_version;
@@ -96,6 +98,11 @@ async fn add_device(
id: format!("gree-{}", normalized_mac.to_ascii_lowercase()),
mac: normalized_mac,
name: input.name.trim().to_string(),
connection_type: ConnectionType::Local,
connection_status: ConnectionStatus::Unknown,
cloud_device_id: None,
cloud_parent_mac: None,
cloud_account_id: None,
ip: input.ip,
port: input.port,
protocol_version: input.protocol_version.min(2),
@@ -126,6 +133,11 @@ async fn add_device(
supports_xfan: None,
supports_health: None,
supports_sleep: None,
supports_buzzer_control: None,
supports_energy_meter: None,
total_energy_kwh: None,
compressor_frequency_hz: None,
last_cloud_sync: None,
current_temperature: if input.simulated { Some(25.0) } else { None },
outdoor_temperature: None,
temperature_sensor_offset: None,
@@ -134,6 +146,13 @@ async fn add_device(
last_seen: if input.simulated { Some(now) } else { None },
last_error: None,
communication_failures: 0,
pending_command: false,
capabilities: crate::models::DeviceCapabilities::default(),
energy_source: crate::models::EnergySourcePreference::Auto,
ha_energy_entity_id: None,
ha_energy_unit: None,
ha_energy_device_class: None,
ha_energy_state_class: None,
created_at: now,
updated_at: now,
};
@@ -173,6 +192,13 @@ async fn patch_device(
.db
.get_device(&id)?
.ok_or_else(|| AppError::NotFound(format!("device {id}")))?;
if device.connection_type == ConnectionType::GreeCloud
&& (patch.ip.is_some() || patch.port.is_some() || patch.protocol_version.is_some() || patch.key.is_some())
{
return Err(AppError::BadRequest(
"IP, UDP port, local protocol and local key are not configurable for GREE Cloud devices".into(),
));
}
if let Some(v) = patch.name {
if !v.trim().is_empty() {
device.name = v.trim().to_string();
@@ -206,6 +232,44 @@ async fn patch_device(
if let Some(v) = patch.enabled {
device.enabled = v;
}
if let Some(v) = patch.energy_source { device.energy_source = v; }
if let Some(v) = patch.ha_energy_entity_id { device.ha_energy_entity_id = v.filter(|x| !x.trim().is_empty()); }
if let Some(v) = patch.ha_energy_unit { device.ha_energy_unit = v.filter(|x| !x.trim().is_empty()); }
if let Some(v) = patch.ha_energy_device_class { device.ha_energy_device_class = v.filter(|x| !x.trim().is_empty()); }
if let Some(v) = patch.ha_energy_state_class { device.ha_energy_state_class = v.filter(|x| !x.trim().is_empty()); }
device.refresh_capabilities();
if device.energy_source == EnergySourcePreference::GreeCloud && !device.capabilities.energy_meter {
return Err(AppError::BadRequest(
"GREE Cloud energy is not available for this device".into(),
));
}
if device.energy_source == EnergySourcePreference::HomeAssistant
&& device.ha_energy_entity_id.as_deref().unwrap_or_default().is_empty()
{
return Err(AppError::BadRequest(
"select a Home Assistant cumulative energy sensor first".into(),
));
}
if device.ha_energy_entity_id.is_some() {
if device.ha_energy_device_class.as_deref() != Some("energy") {
return Err(AppError::BadRequest(
"Home Assistant energy sensor must have device_class=energy".into(),
));
}
if !matches!(device.ha_energy_state_class.as_deref(), Some("total" | "total_increasing")) {
return Err(AppError::BadRequest(
"Home Assistant energy sensor must have state_class=total or total_increasing".into(),
));
}
if !matches!(
device.ha_energy_unit.as_deref().map(str::to_ascii_lowercase).as_deref(),
Some("wh" | "kwh")
) {
return Err(AppError::BadRequest(
"Home Assistant energy sensor must use Wh or kWh".into(),
));
}
}
device.updated_at = Utc::now();
state.db.save_device(&device)?;
state.broadcast("device.updated", serde_json::to_value(&device)?);
@@ -227,13 +291,17 @@ async fn delete_device(
let _house_guard = state.lock_house_operation().await;
let _schedule_guard = state.lock_schedule_operation().await;
let _cycle_guard = state.lock_zone_control_cycle().await;
if state.db.get_device(&id)?.is_none() {
return Err(AppError::NotFound(format!("device {id}")));
}
if state.db.list_automations()?.iter().any(|item| {
item.trigger_device_id.as_deref() == Some(id.as_str())
|| (item.action_group_id.is_none() && item.action_device_id == id)
}) {
let device = state
.db
.get_device(&id)?
.ok_or_else(|| AppError::NotFound(format!("device {id}")))?;
let automations = state.db.list_automations()?;
if device.connection_type == ConnectionType::Local
&& automations.iter().any(|item| {
item.trigger_device_id.as_deref() == Some(id.as_str())
|| (item.action_group_id.is_none() && item.action_device_id == id)
})
{
return Err(AppError::BadRequest(
"device is used by an automation; remove or retarget that automation first".into(),
));
@@ -251,8 +319,40 @@ async fn delete_device(
for zone_id in &sorted_zone_ids {
zone_guards.push(state.lock_zone_operation(zone_id).await);
}
ensure_zone_removal_safe(&state, &removed_zone_ids)?;
ensure_device_stopped_for_detach(&state, &id, "device.deleted").await?;
if device.connection_type == ConnectionType::Local {
ensure_zone_removal_safe(&state, &removed_zone_ids)?;
ensure_device_stopped_for_detach(&state, &id, "device.deleted").await?;
} else {
// Cloud removal must remain possible even when the physical unit is offline. Remove
// controller-only references that would otherwise block deletion, but never send an
// OFF/status request and never depend on MQTT. Local keeps the historical safeguards.
let groups = state.db.list_groups()?;
let emptied_group_ids: std::collections::HashSet<String> = groups
.iter()
.filter(|group| {
!group.zone_ids.is_empty()
&& group.zone_ids.iter().all(|zone_id| removed_zone_ids.contains(zone_id))
})
.map(|group| group.id.clone())
.collect();
for automation in automations.iter().filter(|item| {
item.trigger_device_id.as_deref() == Some(id.as_str())
|| (item.action_group_id.is_none() && item.action_device_id == id)
|| item
.action_group_id
.as_ref()
.is_some_and(|group_id| emptied_group_ids.contains(group_id))
|| item
.action_zone_id
.as_ref()
.is_some_and(|zone_id| removed_zone_ids.contains(zone_id))
}) {
if state.db.delete_automation(&automation.id)? {
state.broadcast("automation.deleted", json!({"id": automation.id.clone()}));
}
}
state.providers.cloud().unregister_device(&id).await;
}
if !state.db.delete_device(&id)? {
return Err(AppError::NotFound(format!("device {id}")));
}
@@ -262,9 +362,10 @@ async fn delete_device(
"info",
"device.deleted",
"Device deleted",
json!({"device_id": id}),
json!({"device_id": id, "connection_type": device.connection_type}),
);
state.broadcast("device.deleted", json!({"id": id}));
state.wake_zone_control();
Ok(StatusCode::NO_CONTENT)
}
@@ -278,11 +379,15 @@ async fn bind_device(
.db
.get_device(&id)?
.ok_or_else(|| AppError::NotFound(format!("device {id}")))?;
if device.connection_type == ConnectionType::GreeCloud {
return Err(AppError::BadRequest("bind is only available for Local/LAN devices".into()));
}
if device.simulated {
return Ok(Json(device));
}
let bound = state
.gree
.providers
.local()
.bind(&device)
.await
.map_err(|e| AppError::Device(e.to_string()))?;
@@ -290,6 +395,7 @@ async fn bind_device(
device.protocol_version = bound.protocol_version;
device.communication_failures = 0;
device.online = true;
device.connection_status = ConnectionStatus::Online;
device.last_seen = Some(Utc::now());
device.last_error = None;
device.updated_at = Utc::now();
@@ -319,8 +425,13 @@ async fn probe_device(
.db
.get_device(&id)?
.ok_or_else(|| AppError::NotFound(format!("device {id}")))?;
if device.connection_type == ConnectionType::GreeCloud {
return Err(AppError::BadRequest("UDP probe is only available for Local/LAN devices".into()));
}
let response_time_ms = state
.gree
.providers
.local()
.client()
.probe(&device)
.await
.map_err(|err| AppError::Device(err.to_string()))?;
+418
View File
@@ -0,0 +1,418 @@
fn cloud_debug_event(state: &AppState, data: Value) {
if state.debug_cloud_requests.load(Ordering::Relaxed) {
state.broadcast_with_control_plan_invalidation("gree_cloud.request", data, false);
}
}
fn cloud_error_kind(error: &anyhow::Error) -> &'static str {
let text = format!("{error:#}").to_ascii_lowercase();
if text.contains("authentication failed") || text.contains("login failed") {
"authentication_error"
} else if text.contains("timeout") || text.contains("timed out") {
"timeout"
} else if text.contains("http 5") || text.contains("service unavailable") {
"api_unavailable"
} else if text.contains("connect") || text.contains("dns") || text.contains("network") {
"network_error"
} else {
"api_error"
}
}
async fn cloud_api_from_settings(state: &AppState) -> Result<crate::protocol::gree_cloud::GreeCloudApi, AppError> {
let settings = state.settings.read().await.gree_cloud.clone();
crate::protocol::gree_cloud::GreeCloudApi::for_region(
state.http.clone(),
&settings.region,
&settings.username,
&settings.password,
)
.map_err(|err| AppError::BadRequest(err.to_string()))
}
async fn test_gree_cloud(State(state): State<AppState>) -> Result<Json<Value>, AppError> {
let mut api = cloud_api_from_settings(&state).await?;
let started = Instant::now();
cloud_debug_event(&state, json!({"operation":"test_connection","phase":"sent"}));
let result = async {
api.login().await?;
let devices = api.get_all_devices().await?;
Ok::<usize, anyhow::Error>(devices.len())
}
.await;
match result {
Ok(device_count) => {
let now = Utc::now();
cloud_debug_event(&state, json!({
"operation":"test_connection",
"phase":"response",
"duration_ms": started.elapsed().as_millis() as u64,
"device_count": device_count,
}));
{
let _configuration_guard = state.lock_configuration_operation().await;
let mut settings = state.settings.write().await;
settings.gree_cloud.last_successful_contact = Some(now);
settings.gree_cloud.last_rest_response_time_ms =
Some(started.elapsed().as_millis().min(u64::MAX as u128) as u64);
state.db.save_runtime_settings(&settings)?;
}
state.log(
"info",
"gree_cloud.login_success",
"GREE Cloud connection test succeeded",
json!({"device_count": device_count}),
);
Ok(Json(json!({
"ok": true,
"status": "connected",
"device_count": device_count,
"last_successful_contact": now,
})))
}
Err(error) => {
let kind = cloud_error_kind(&error);
cloud_debug_event(&state, json!({
"operation":"test_connection",
"phase":"error",
"duration_ms": started.elapsed().as_millis() as u64,
"kind": kind,
}));
tracing::warn!(kind, "GREE Cloud connection test failed");
state.log(
"warn",
"gree_cloud.login_failure",
"GREE Cloud connection test failed",
json!({"kind": kind}),
);
Ok(Json(json!({
"ok": false,
"status": kind,
"message": match kind {
"authentication_error" => "Invalid GREE Cloud login/password or authorization was rejected",
"timeout" => "GREE Cloud request timed out",
"network_error" => "Cannot reach GREE Cloud",
"api_unavailable" => "GREE Cloud API is temporarily unavailable",
_ => "GREE Cloud returned an unexpected response",
}
})))
}
}
}
async fn discover_gree_cloud_devices(
State(state): State<AppState>,
) -> Result<Json<Value>, AppError> {
let mut api = cloud_api_from_settings(&state).await?;
let started = Instant::now();
cloud_debug_event(&state, json!({"operation":"discovery","phase":"sent"}));
api.login()
.await
.map_err(|err| {
cloud_debug_event(&state, json!({
"operation":"discovery",
"phase":"error",
"stage":"login",
"duration_ms": started.elapsed().as_millis() as u64,
"kind": cloud_error_kind(&err),
}));
AppError::Dependency(format!("GREE Cloud login failed: {err}"))
})?;
let devices = api
.get_all_devices()
.await
.map_err(|err| {
cloud_debug_event(&state, json!({
"operation":"discovery",
"phase":"error",
"stage":"devices",
"duration_ms": started.elapsed().as_millis() as u64,
"kind": cloud_error_kind(&err),
}));
AppError::Dependency(format!("GREE Cloud discovery failed: {err}"))
})?;
let rest_duration_ms = started.elapsed().as_millis().min(u64::MAX as u128) as u64;
cloud_debug_event(&state, json!({
"operation":"discovery",
"phase":"response",
"duration_ms": rest_duration_ms,
"device_count": devices.len(),
}));
{
let _configuration_guard = state.lock_configuration_operation().await;
let mut settings = state.settings.write().await;
settings.gree_cloud.last_successful_contact = Some(Utc::now());
settings.gree_cloud.last_rest_response_time_ms = Some(rest_duration_ms);
state.db.save_runtime_settings(&settings)?;
}
let existing = state.db.list_devices()?;
let views = devices
.into_iter()
.map(|device| {
let id = device.mac.replace([':', '-'], "").to_ascii_uppercase();
let already_added = existing.iter().any(|saved| {
saved.connection_type == ConnectionType::GreeCloud
&& saved.cloud_device_id.as_deref().is_some_and(|value| value.eq_ignore_ascii_case(&id))
});
crate::protocol::gree_cloud::CloudDeviceView {
parent_mac: crate::protocol::gree_cloud::parent_mac(&id),
id: id.clone(),
name: device.name,
mac: id,
model: device.model,
version: device.version,
online: device.online,
already_added,
}
})
.collect::<Vec<_>>();
state.log(
"info",
"gree_cloud.discovery",
&format!("GREE Cloud discovery found {} device(s)", views.len()),
json!({"count": views.len()}),
);
Ok(Json(json!({"count": views.len(), "devices": views})))
}
async fn add_gree_cloud_device(
State(state): State<AppState>,
Path(cloud_id): Path<String>,
) -> Result<(StatusCode, Json<Device>), AppError> {
let cloud_id = cloud_id.replace([':', '-'], "").to_ascii_uppercase();
if cloud_id.is_empty() {
return Err(AppError::BadRequest("cloud device id is required".into()));
}
if state.db.list_devices()?.iter().any(|device| {
device.connection_type == ConnectionType::GreeCloud
&& device.cloud_device_id.as_deref() == Some(cloud_id.as_str())
}) {
return Err(AppError::Conflict(
"this GREE Cloud device is already added".into(),
));
}
// Re-discover server-side so the frontend never needs to submit/store the device cipher key.
let mut api = cloud_api_from_settings(&state).await?;
let started = Instant::now();
cloud_debug_event(&state, json!({
"operation":"add_device_lookup",
"phase":"sent",
"cloud_device_id": cloud_id.clone(),
}));
api.login()
.await
.map_err(|err| {
cloud_debug_event(&state, json!({
"operation":"add_device_lookup",
"phase":"error",
"stage":"login",
"duration_ms": started.elapsed().as_millis() as u64,
"kind": cloud_error_kind(&err),
}));
AppError::Dependency(format!("GREE Cloud login failed: {err}"))
})?;
let cloud_device = api
.get_all_devices()
.await
.map_err(|err| {
cloud_debug_event(&state, json!({
"operation":"add_device_lookup",
"phase":"error",
"stage":"devices",
"duration_ms": started.elapsed().as_millis() as u64,
"kind": cloud_error_kind(&err),
}));
AppError::Dependency(format!("GREE Cloud discovery failed: {err}"))
})?
.into_iter()
.find(|device| device.mac.eq_ignore_ascii_case(&cloud_id))
.ok_or_else(|| AppError::NotFound(format!("GREE Cloud device {cloud_id}")))?;
cloud_debug_event(&state, json!({
"operation":"add_device_lookup",
"phase":"response",
"duration_ms": started.elapsed().as_millis() as u64,
"cloud_device_id": cloud_id.clone(),
}));
let _configuration_guard = state.lock_configuration_operation().await;
let now = Utc::now();
let account_id = state.settings.read().await.gree_cloud.account_id.clone();
let normalized_cloud_mac = cloud_device.mac.replace([':', '-'], "").to_ascii_uppercase();
let device = Device {
id: format!("gree-cloud-{}", normalized_cloud_mac.to_ascii_lowercase()),
mac: normalized_cloud_mac.clone(),
name: if cloud_device.name.trim().is_empty() {
format!("GREE Cloud {}", &normalized_cloud_mac)
} else {
cloud_device.name.clone()
},
connection_type: ConnectionType::GreeCloud,
connection_status: ConnectionStatus::CloudDisconnected,
cloud_device_id: Some(normalized_cloud_mac.clone()),
cloud_parent_mac: Some(crate::protocol::gree_cloud::parent_mac(&normalized_cloud_mac)),
cloud_account_id: Some(account_id),
ip: String::new(),
port: 0,
// The reference HA integration currently creates CloudDevice with cipher_version=1.
protocol_version: 1,
model: cloud_device.model.unwrap_or_default(),
firmware: cloud_device.version.unwrap_or_default(),
key: Some(cloud_device.key),
cid: Some("gree-cloud".into()),
enabled: true,
simulated: false,
power: false,
mode: "cool".into(),
target_temperature: 24.0,
fan_speed: 0,
swing_vertical: false,
swing_horizontal: false,
quiet: false,
quiet_wire_value: None,
turbo: false,
light: false,
air: false,
xfan: false,
health: false,
sleep: false,
supports_light: None,
supports_quiet: None,
supports_turbo: None,
supports_air: None,
supports_xfan: None,
supports_health: None,
supports_sleep: None,
supports_buzzer_control: None,
supports_energy_meter: None,
total_energy_kwh: None,
compressor_frequency_hz: None,
last_cloud_sync: None,
current_temperature: None,
outdoor_temperature: None,
temperature_sensor_offset: None,
online: false,
response_time_ms: None,
last_seen: None,
last_error: None,
communication_failures: 0,
pending_command: false,
capabilities: crate::models::DeviceCapabilities {
vertical_swing: false,
horizontal_swing: false,
..crate::models::DeviceCapabilities::default()
},
energy_source: crate::models::EnergySourcePreference::Auto,
ha_energy_entity_id: None,
ha_energy_unit: None,
ha_energy_device_class: None,
ha_energy_state_class: None,
created_at: now,
updated_at: now,
};
state.db.save_device(&device)?;
state.log(
"info",
"gree_cloud.device_added",
&format!("Added GREE Cloud device {}", device.name),
json!({"device_id": device.id, "cloud_device_id": device.cloud_device_id}),
);
state.broadcast("device.created", serde_json::to_value(&device)?);
Ok((StatusCode::CREATED, Json(device)))
}
async fn gree_cloud_status(State(state): State<AppState>) -> Result<Json<Value>, AppError> {
let settings = state.settings.read().await.gree_cloud.clone();
let mqtt_connected = state.providers.cloud().is_connected().await;
let cloud_devices = state
.db
.list_devices()?
.into_iter()
.filter(|device| device.connection_type == ConnectionType::GreeCloud)
.collect::<Vec<_>>();
let account_status = if !settings.enabled {
"disabled"
} else if settings.username.trim().is_empty() || settings.password.trim().is_empty() {
"not_configured"
} else if cloud_devices
.iter()
.any(|device| device.connection_status == ConnectionStatus::AuthenticationError)
{
"authentication_error"
} else if mqtt_connected {
"connected"
} else {
"cloud_disconnected"
};
let runtime = state.providers.cloud().runtime_status().await;
Ok(Json(json!({
"enabled": settings.enabled,
"account_status": account_status,
"mqtt_status": if mqtt_connected { "connected" } else { "disconnected" },
"last_successful_contact": settings.last_successful_contact,
"last_rest_response_time_ms": settings.last_rest_response_time_ms,
"device_count": cloud_devices.len(),
"online_device_count": cloud_devices.iter().filter(|device| device.connection_status == ConnectionStatus::Online).count(),
"runtime": runtime,
})))
}
async fn reconnect_gree_cloud(State(state): State<AppState>) -> Result<Json<Value>, AppError> {
let settings = state.settings.read().await.gree_cloud.clone();
if !settings.enabled {
return Err(AppError::BadRequest("GREE Cloud is disabled".into()));
}
let devices = state.db.list_devices()?;
state.providers.cloud().shutdown().await;
state
.providers
.cloud()
.ensure_connected(&settings, &devices)
.await
.map_err(|error| AppError::Dependency(cloud_public_error_text(&error.to_string())))?;
let now = Utc::now();
{
let mut runtime = state.settings.write().await;
runtime.gree_cloud.last_successful_contact = Some(now);
state.db.save_runtime_settings(&runtime)?;
}
state.log(
"info",
"gree_cloud.reconnect",
"GREE Cloud MQTT reconnected",
json!({"device_count": devices.iter().filter(|device| device.connection_type == ConnectionType::GreeCloud).count()}),
);
Ok(Json(json!({"ok": true, "mqtt_status": "connected", "last_successful_contact": now})))
}
async fn cloud_device_diagnostics(
State(state): State<AppState>,
Path(id): Path<String>,
) -> Result<Json<Value>, AppError> {
let device = state
.db
.get_device(&id)?
.ok_or_else(|| AppError::NotFound(format!("device {id}")))?;
if device.connection_type != ConnectionType::GreeCloud {
return Err(AppError::BadRequest("Cloud diagnostics are available only for GREE Cloud devices".into()));
}
let diagnostics = state.providers.cloud().diagnostics(&device.id).await;
Ok(Json(json!({
"device_id": device.id,
"cloud_device_id": device.cloud_device_id,
"connection_status": device.connection_status,
"last_sync": device.last_cloud_sync,
"capabilities": device.capabilities,
"provider": diagnostics,
})))
}
fn cloud_public_error_text(error: &str) -> String {
let lower = error.to_ascii_lowercase();
if lower.contains("password") || lower.contains("token") || lower.contains("authorization") {
"GREE Cloud authentication failed".into()
} else {
error.chars().take(300).collect()
}
}
+211
View File
@@ -488,3 +488,214 @@ async fn control_plan(State(state): State<AppState>) -> Result<Json<Value>, AppE
let snapshot = engine::get_control_plan_snapshot(&state).await?;
Ok(Json(serde_json::to_value(snapshot.plan.as_ref())?))
}
#[derive(Debug, Deserialize)]
struct EnergyHistoryQuery {
device_id: String,
interval: Option<String>,
source: Option<String>,
days: Option<i64>,
limit: Option<u32>,
}
async fn energy_history(
State(state): State<AppState>,
Query(query): Query<EnergyHistoryQuery>,
) -> Result<Json<Value>, AppError> {
use chrono::{Datelike, NaiveDate, Timelike, Weekday};
use std::collections::BTreeMap;
let device = state
.db
.get_device(&query.device_id)?
.ok_or_else(|| AppError::NotFound(format!("device {}", query.device_id)))?;
let interval_name = query.interval.as_deref().unwrap_or("daily");
if !matches!(interval_name, "hourly" | "daily" | "weekly" | "monthly") {
return Err(AppError::BadRequest(
"energy interval must be hourly, daily, weekly or monthly".into(),
));
}
let days = query.days.unwrap_or(31).clamp(1, 3650);
let now = Utc::now();
let since = now - ChronoDuration::days(days);
let month_start_for_load = chrono::DateTime::<Utc>::from_naive_utc_and_offset(
chrono::NaiveDate::from_ymd_opt(now.year(), now.month(), 1)
.expect("valid current month")
.and_hms_opt(0, 0, 0)
.expect("valid midnight"),
Utc,
);
let previous_month_date_for_load = month_start_for_load.date_naive() - ChronoDuration::days(1);
let previous_month_start_for_load = chrono::DateTime::<Utc>::from_naive_utc_and_offset(
chrono::NaiveDate::from_ymd_opt(
previous_month_date_for_load.year(),
previous_month_date_for_load.month(),
1,
)
.expect("valid previous month")
.and_hms_opt(0, 0, 0)
.expect("valid midnight"),
Utc,
);
let load_since = since.min(previous_month_start_for_load);
let limit = query.limit.unwrap_or(100_000).clamp(1, 200_000);
let influx = state.settings.read().await.influxdb.clone();
let cutoff = now - ChronoDuration::days(influx.history_threshold_days as i64);
let mut storage = "sqlite".to_string();
let mut storage_warning: Option<String> = None;
let mut samples = if influx.enabled && load_since < cutoff {
match influxdb::query_energy(
&state.http,
&influx,
&device.id,
None,
load_since,
cutoff,
3600,
limit,
)
.await
{
Ok(mut archived) => {
archived.extend(state.db.list_energy_readings(&device.id, cutoff, limit)?);
storage = "influx+sqlite".into();
archived
}
Err(err) => {
storage = "sqlite_fallback".into();
storage_warning = Some(err.to_string());
state.log(
"warn",
"influx.query_error",
"InfluxDB energy history query failed",
json!({"device_id": device.id, "error": err.to_string()}),
);
state.db.list_energy_readings(&device.id, load_since, limit)?
}
}
} else {
state.db.list_energy_readings(&device.id, load_since, limit)?
};
samples.sort_by_key(|row| row.timestamp);
let requested_source = query.source.as_deref().unwrap_or("auto");
let selected_source = match requested_source {
"gree_cloud" => Some("gree_cloud"),
"home_assistant" => Some("home_assistant"),
"auto" => match device.energy_source {
EnergySourcePreference::GreeCloud => Some("gree_cloud"),
EnergySourcePreference::HomeAssistant => Some("home_assistant"),
EnergySourcePreference::Auto => {
if samples.iter().any(|row| row.source == "gree_cloud") {
Some("gree_cloud")
} else if samples.iter().any(|row| row.source == "home_assistant") {
Some("home_assistant")
} else {
None
}
}
},
_ => {
return Err(AppError::BadRequest(
"energy source must be auto, gree_cloud or home_assistant".into(),
))
}
};
if let Some(source) = selected_source {
samples.retain(|row| row.source == source);
} else {
samples.clear();
}
fn midnight(date: NaiveDate) -> chrono::DateTime<Utc> {
chrono::DateTime::<Utc>::from_naive_utc_and_offset(
date.and_hms_opt(0, 0, 0).expect("valid midnight"),
Utc,
)
}
fn bucket_start(
timestamp: chrono::DateTime<Utc>,
interval_name: &str,
) -> chrono::DateTime<Utc> {
let date = timestamp.date_naive();
match interval_name {
"hourly" => chrono::DateTime::<Utc>::from_naive_utc_and_offset(
date.and_hms_opt(timestamp.hour(), 0, 0)
.expect("valid hour"),
Utc,
),
"weekly" => {
let iso = date.iso_week();
midnight(
NaiveDate::from_isoywd_opt(iso.year(), iso.week(), Weekday::Mon)
.expect("valid ISO week"),
)
}
"monthly" => midnight(
NaiveDate::from_ymd_opt(date.year(), date.month(), 1)
.expect("valid month"),
),
_ => midnight(date),
}
}
let period_samples = samples
.iter()
.filter(|row| row.timestamp >= since)
.collect::<Vec<_>>();
let mut buckets: BTreeMap<chrono::DateTime<Utc>, f64> = BTreeMap::new();
for sample in &period_samples {
*buckets
.entry(bucket_start(sample.timestamp, interval_name))
.or_default() += sample.consumption_kwh.max(0.0);
}
let buckets = buckets
.into_iter()
.map(|(start, consumption_kwh)| {
json!({"start": start, "consumption_kwh": consumption_kwh.max(0.0)})
})
.collect::<Vec<_>>();
let today_start = midnight(now.date_naive());
let yesterday_start = today_start - ChronoDuration::days(1);
let month_start = midnight(
NaiveDate::from_ymd_opt(now.year(), now.month(), 1).expect("valid current month"),
);
let previous_month_date = month_start.date_naive() - ChronoDuration::days(1);
let previous_month_start = midnight(
NaiveDate::from_ymd_opt(previous_month_date.year(), previous_month_date.month(), 1)
.expect("valid previous month"),
);
let sum_range = |start: chrono::DateTime<Utc>, stop: chrono::DateTime<Utc>| -> f64 {
samples
.iter()
.filter(|row| row.timestamp >= start && row.timestamp < stop)
.map(|row| row.consumption_kwh.max(0.0))
.sum()
};
let period_total: f64 = period_samples
.iter()
.map(|row| row.consumption_kwh.max(0.0))
.sum();
let latest = samples.last().cloned();
Ok(Json(json!({
"device_id": device.id,
"source": selected_source.unwrap_or("none"),
"configured_source": device.energy_source,
"interval": interval_name,
"unit": "kWh",
"period_days": days,
"storage": storage,
"storage_warning": storage_warning,
"buckets": buckets,
"summary": {
"today": sum_range(today_start, now + ChronoDuration::seconds(1)),
"yesterday": sum_range(yesterday_start, today_start),
"current_month": sum_range(month_start, now + ChronoDuration::seconds(1)),
"previous_month": sum_range(previous_month_start, month_start),
"period_total": period_total,
},
"latest": latest,
})))
}
+33
View File
@@ -95,3 +95,36 @@ async fn test_notifications(
.map_err(AppError::Device)?;
Ok(Json(json!({"ok": true})))
}
async fn list_home_assistant_energy_sensors(
State(state): State<AppState>,
) -> Result<Json<Value>, AppError> {
let settings = state.settings.read().await.home_assistant.clone();
let entities = home_assistant::list_entities(&state.http, &settings)
.await
.map_err(|error| AppError::Device(error.to_string()))?;
let sensors = entities
.into_iter()
.filter_map(|entity| {
let attributes = entity.get("attributes")?.as_object()?;
let device_class = attributes.get("device_class")?.as_str()?;
let state_class = attributes.get("state_class")?.as_str()?;
let unit = attributes.get("unit_of_measurement")?.as_str()?;
if device_class != "energy"
|| !matches!(state_class, "total" | "total_increasing")
|| !matches!(unit.to_ascii_lowercase().as_str(), "wh" | "kwh")
{
return None;
}
Some(json!({
"entity_id": entity.get("entity_id").and_then(Value::as_str).unwrap_or_default(),
"name": attributes.get("friendly_name").and_then(Value::as_str).unwrap_or_default(),
"state": entity.get("state").and_then(Value::as_str).unwrap_or_default(),
"unit": unit,
"device_class": device_class,
"state_class": state_class,
}))
})
.collect::<Vec<_>>();
Ok(Json(json!({"sensors": sensors})))
}
+99
View File
@@ -17,6 +17,20 @@ fn gree_settings(settings: &RuntimeSettings) -> GreeSettings {
}
}
fn gree_cloud_settings(settings: &RuntimeSettings) -> GreeCloudSettingsView {
GreeCloudSettingsView {
enabled: settings.gree_cloud.enabled,
region: settings.gree_cloud.region.clone(),
username: settings.gree_cloud.username.clone(),
password_configured: !settings.gree_cloud.password.is_empty(),
polling_interval_seconds: settings.gree_cloud.polling_interval_seconds,
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,
last_rest_response_time_ms: settings.gree_cloud.last_rest_response_time_ms,
}
}
fn history_settings(settings: &RuntimeSettings) -> HistorySettings {
HistorySettings {
retention_days: settings.history_retention_days,
@@ -74,6 +88,7 @@ fn settings_snapshot(settings: &RuntimeSettings) -> SettingsSnapshot {
SettingsSnapshot {
application: application_settings(settings),
gree: gree_settings(settings),
gree_cloud: gree_cloud_settings(settings),
history: history_settings(settings),
influxdb: influxdb_settings(settings),
notifications: notification_settings(settings),
@@ -209,6 +224,84 @@ async fn update_gree_settings(
Ok(Json(payload))
}
async fn get_gree_cloud_settings(
State(state): State<AppState>,
) -> Json<GreeCloudSettingsView> {
Json(gree_cloud_settings(&*state.settings.read().await))
}
fn apply_gree_cloud_update(
current: &GreeCloudSettings,
input: GreeCloudSettingsUpdate,
) -> Result<GreeCloudSettings, AppError> {
if crate::protocol::gree_cloud::region_base_url(input.region.trim()).is_none() {
return Err(AppError::BadRequest(format!(
"unsupported GREE Cloud region: {}",
input.region
)));
}
let mut next = current.clone();
next.enabled = input.enabled;
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);
if let Some(password) = input.password {
next.password = password;
}
if next.enabled && next.username.is_empty() {
return Err(AppError::BadRequest(
"GREE Cloud login/email is required when cloud is enabled".into(),
));
}
if next.enabled && next.password.is_empty() {
return Err(AppError::BadRequest(
"GREE Cloud password is required when cloud is enabled".into(),
));
}
Ok(next)
}
async fn update_gree_cloud_settings(
State(state): State<AppState>,
Json(input): Json<GreeCloudSettingsUpdate>,
) -> Result<Json<GreeCloudSettingsView>, AppError> {
let _configuration_guard = state.lock_configuration_operation().await;
let (payload, reconnect_required) = {
let mut settings = state.settings.write().await;
let previous = settings.gree_cloud.clone();
let next = apply_gree_cloud_update(&previous, input)?;
let reconnect_required = previous.enabled != next.enabled
|| previous.region != next.region
|| previous.username != next.username
|| previous.password != next.password;
settings.gree_cloud = next;
state.db.save_runtime_settings(&settings)?;
(gree_cloud_settings(&settings), reconnect_required)
};
if reconnect_required {
// Never keep a broker session authenticated with stale/disabled account settings.
// The reconnect loop will establish a fresh session when Cloud remains enabled.
state.providers.cloud().shutdown().await;
}
state.log(
"info",
"settings.gree_cloud.updated",
"GREE Cloud settings updated",
json!({
"enabled": payload.enabled,
"region": payload.region,
"polling_interval_seconds": payload.polling_interval_seconds,
"password_configured": payload.password_configured
}),
);
state.broadcast(
"settings.gree_cloud.updated",
serde_json::to_value(&payload)?,
);
Ok(Json(payload))
}
async fn get_history_settings(State(state): State<AppState>) -> Json<HistorySettings> {
Json(history_settings(&*state.settings.read().await))
}
@@ -645,6 +738,12 @@ async fn update_debug_settings(
state
.debug_gree_frames
.store(input.gree_frames, Ordering::Relaxed);
state
.debug_cloud_requests
.store(input.cloud_requests, Ordering::Relaxed);
state
.debug_cloud_mqtt
.store(input.cloud_mqtt, Ordering::Relaxed);
state.broadcast("settings.debug.updated", serde_json::to_value(&input)?);
Ok(Json(input))
}
+1 -1
View File
@@ -79,7 +79,7 @@ async fn build_bootstrap(state: &AppState) -> Result<BootstrapResponse, AppError
}
fn build_system_info(state: &AppState, devices: &[Device]) -> SystemInfoResponse {
let (received_frames_total, received_frames_by_device) = state.gree.received_frame_stats();
let (received_frames_total, received_frames_by_device) = state.providers.local().client().received_frame_stats();
SystemInfoResponse {
version: env!("CARGO_PKG_VERSION"),
uptime_seconds: state.started.elapsed().as_secs(),
+10 -1
View File
@@ -1,5 +1,5 @@
use crate::models::{
DebugSettings, HomeAssistantSettings, InfluxDbSettings, NightModeSettings,
DebugSettings, GreeCloudSettings, HomeAssistantSettings, InfluxDbSettings, NightModeSettings,
NotificationSettings, RuntimeSettings,
};
use anyhow::{Context, Result};
@@ -100,8 +100,11 @@ impl Config {
debug: DebugSettings {
overlay_enabled: env_bool("GREE_CONTROLLER_DEBUG_OVERLAY").unwrap_or(false),
gree_frames: env_bool("GREE_CONTROLLER_DEBUG_GREE_FRAMES").unwrap_or(false),
cloud_requests: env_bool("GREE_CONTROLLER_DEBUG_CLOUD_REQUESTS").unwrap_or(false),
cloud_mqtt: env_bool("GREE_CONTROLLER_DEBUG_CLOUD_MQTT").unwrap_or(false),
},
notifications: NotificationSettings::default(),
gree_cloud: GreeCloudSettings::default(),
night_mode: NightModeSettings {
enabled: env_bool("GREE_CONTROLLER_NIGHT_MODE_ENABLED").unwrap_or(false),
start_time: env::var("GREE_CONTROLLER_NIGHT_MODE_START")
@@ -162,6 +165,12 @@ impl Config {
if let Some(value) = env_bool("GREE_CONTROLLER_DEBUG_GREE_FRAMES") {
settings.debug.gree_frames = value;
}
if let Some(value) = env_bool("GREE_CONTROLLER_DEBUG_CLOUD_REQUESTS") {
settings.debug.cloud_requests = value;
}
if let Some(value) = env_bool("GREE_CONTROLLER_DEBUG_CLOUD_MQTT") {
settings.debug.cloud_mqtt = value;
}
if let Some(value) = env_bool("GREE_CONTROLLER_NIGHT_MODE_ENABLED") {
settings.night_mode.enabled = value;
}
+3 -2
View File
@@ -1,7 +1,7 @@
use crate::{
models::{
ApiTokenInfo, Automation, ClimateGroup, ConfigurationExport, Device, EventLog, Flow,
HaReading, Reading, RuntimeSettings, Schedule, Zone, ZoneReading,
ApiTokenInfo, Automation, ClimateGroup, ConfigurationExport, ConnectionType, Device, EventLog, Flow,
EnergyReading, HaReading, Reading, RuntimeSettings, Schedule, Zone, ZoneReading,
},
queries,
};
@@ -28,6 +28,7 @@ include!("db/flows.rs");
include!("db/device_history.rs");
include!("db/zone_history.rs");
include!("db/ha_history.rs");
include!("db/energy_history.rs");
include!("db/events_tokens.rs");
include!("db/configuration.rs");
include!("db/tests.rs");
+6 -2
View File
@@ -18,12 +18,16 @@ impl Db {
let tx = conn.transaction()?;
tx.execute_batch(queries::CLEAR_CONFIGURATION)?;
for device in &export.devices {
let payload = Self::to_json(device)?;
let payload = Self::device_to_storage_json(device)?;
let index_mac = match device.connection_type {
ConnectionType::Local => device.mac.clone(),
ConnectionType::GreeCloud => format!("cloud:{}", device.cloud_device_id.as_deref().unwrap_or(&device.mac)),
};
tx.execute(
queries::UPSERT_DEVICE,
params![
device.id,
device.mac,
index_mac,
device.name,
device.ip,
device.simulated as i64,
+19 -2
View File
@@ -23,6 +23,18 @@ impl Db {
Ok(serde_json::to_string(value)?)
}
/// Device protocol keys are intentionally omitted by the public `Device` serializer.
/// Persist them only in the private SQLite payload so normal API responses never expose them.
fn device_to_storage_json(device: &Device) -> Result<String> {
let mut value = serde_json::to_value(device)?;
if let Some(object) = value.as_object_mut() {
if let Some(key) = device.key.as_ref().filter(|key| !key.is_empty()) {
object.insert("key".into(), serde_json::Value::String(key.clone()));
}
}
Ok(serde_json::to_string(&value)?)
}
pub fn count_devices(&self) -> Result<u64> {
let conn = self.lock()?;
let count: i64 = conn.query_row(queries::COUNT_DEVICES, [], |row| row.get(0))?;
@@ -30,13 +42,17 @@ impl Db {
}
pub fn save_device(&self, device: &Device) -> Result<()> {
let payload = Self::to_json(device)?;
let payload = Self::device_to_storage_json(device)?;
let index_mac = match device.connection_type {
ConnectionType::Local => device.mac.clone(),
ConnectionType::GreeCloud => format!("cloud:{}", device.cloud_device_id.as_deref().unwrap_or(&device.mac)),
};
let conn = self.lock()?;
conn.execute(
queries::UPSERT_DEVICE,
params![
device.id,
device.mac,
index_mac,
device.name,
device.ip,
device.simulated as i64,
@@ -76,6 +92,7 @@ impl Db {
let mut conn = self.lock()?;
let tx = conn.transaction()?;
tx.execute(queries::DELETE_DEVICE_READINGS, [id])?;
tx.execute(queries::DELETE_DEVICE_ENERGY_READINGS, [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])?;
+95
View File
@@ -0,0 +1,95 @@
impl Db {
pub fn add_energy_reading(&self, reading: &EnergyReading) -> Result<i64> {
let conn = self.lock()?;
conn.execute(
"INSERT INTO energy_readings(device_id,timestamp,source,raw_meter_value,raw_unit,normalized_meter_kwh,consumption_kwh,current_power_kw,quality,reset_detected) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)",
params![
reading.device_id,
reading.timestamp.to_rfc3339(),
reading.source,
reading.raw_meter_value,
reading.raw_unit,
reading.normalized_meter_kwh,
reading.consumption_kwh.max(0.0),
reading.current_power_kw,
reading.quality,
reading.reset_detected as i64,
],
)?;
Ok(conn.last_insert_rowid())
}
pub fn last_energy_reading(&self, device_id: &str, source: &str) -> Result<Option<EnergyReading>> {
let conn = self.lock()?;
conn.query_row(
"SELECT id,device_id,timestamp,source,raw_meter_value,raw_unit,normalized_meter_kwh,consumption_kwh,current_power_kw,quality,reset_detected FROM energy_readings WHERE device_id=?1 AND source=?2 ORDER BY timestamp DESC,id DESC LIMIT 1",
params![device_id, source],
Self::map_energy_reading,
).optional().map_err(Into::into)
}
pub fn list_energy_readings(
&self,
device_id: &str,
since: DateTime<Utc>,
limit: u32,
) -> Result<Vec<EnergyReading>> {
let conn = self.lock()?;
let mut stmt = conn.prepare(
"SELECT id,device_id,timestamp,source,raw_meter_value,raw_unit,normalized_meter_kwh,consumption_kwh,current_power_kw,quality,reset_detected FROM energy_readings WHERE device_id=?1 AND timestamp>=?2 ORDER BY timestamp ASC,id ASC LIMIT ?3",
)?;
let rows = stmt.query_map(
params![device_id, since.to_rfc3339(), limit.clamp(1, 100_000) as i64],
Self::map_energy_reading,
)?;
rows.collect::<rusqlite::Result<Vec<_>>>().map_err(Into::into)
}
pub fn energy_before(&self, before: DateTime<Utc>, limit: u32) -> Result<Vec<EnergyReading>> {
let conn = self.lock()?;
let mut stmt = conn.prepare(
"SELECT id,device_id,timestamp,source,raw_meter_value,raw_unit,normalized_meter_kwh,consumption_kwh,current_power_kw,quality,reset_detected FROM energy_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_energy_reading,
)?;
rows.collect::<rusqlite::Result<Vec<_>>>().map_err(Into::into)
}
pub fn delete_energy_batch(&self, rows: &[EnergyReading]) -> 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 energy_readings WHERE id=?1", [row.id])? as u64;
}
tx.commit()?;
Ok(changed)
}
pub fn prune_energy_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 energy_readings WHERE timestamp < ?1", [before.to_rfc3339()])? as u64)
}
fn map_energy_reading(row: &rusqlite::Row<'_>) -> rusqlite::Result<EnergyReading> {
let timestamp: String = row.get(2)?;
Ok(EnergyReading {
id: row.get(0)?,
device_id: row.get(1)?,
timestamp: DateTime::parse_from_rfc3339(&timestamp)
.map(|value| value.with_timezone(&Utc))
.unwrap_or_else(|_| Utc::now()),
source: row.get(3)?,
raw_meter_value: row.get(4)?,
raw_unit: row.get(5)?,
normalized_meter_kwh: row.get(6)?,
consumption_kwh: row.get::<_, f64>(7)?.max(0.0),
current_power_kw: row.get(8)?,
quality: row.get(9)?,
reset_detected: row.get::<_, i64>(10)? != 0,
})
}
}
+25 -1
View File
@@ -1,7 +1,7 @@
#[cfg(test)]
mod tests {
use super::*;
use crate::models::{ApiTokenInfo, Device, HaReading, Reading};
use crate::models::{ApiTokenInfo, ConnectionType, Device, HaReading, Reading};
#[test]
fn history_compaction_keeps_one_sample_per_old_bucket() {
@@ -129,4 +129,28 @@ mod tests {
);
assert_eq!(db.history_counts().unwrap(), (1, 0, 1));
}
#[test]
fn local_and_cloud_entries_with_same_physical_mac_can_coexist() {
let dir = tempfile::tempdir().unwrap();
let db = Db::open(&dir.path().join("duplicate-transport.db")).unwrap();
let mut local = Device::simulated_default();
local.id = "local-device".into();
local.mac = "AABBCCDDEEFF".into();
local.connection_type = ConnectionType::Local;
let mut cloud = local.clone();
cloud.id = "cloud-device".into();
cloud.connection_type = ConnectionType::GreeCloud;
cloud.cloud_device_id = Some(local.mac.clone());
cloud.ip.clear();
cloud.port = 0;
cloud.key = Some("0123456789abcdef".into());
db.save_device(&local).unwrap();
db.save_device(&cloud).unwrap();
let devices = db.list_devices().unwrap();
assert_eq!(devices.len(), 2);
assert!(devices.iter().any(|item| item.connection_type == ConnectionType::Local));
assert!(devices.iter().any(|item| item.connection_type == ConnectionType::GreeCloud));
}
}
+3 -2
View File
@@ -2,7 +2,7 @@ use crate::{
error::AppError,
home_assistant, influxdb,
models::{
Automation, AutomationPlanRule, ClimateGroup, ControlPlan, ControlPlanEvent, Device,
Automation, AutomationPlanRule, ClimateGroup, ConnectionStatus, ConnectionType, ControlPlan, ControlPlanEvent, Device, EnergyReading,
DeviceCommand, GroupControlPatch, HaReading, NightModeSettings, Reading, RuntimeSettings,
Schedule, TemporaryQuickThermostat, Zone, ZoneControlPlan, ZoneReading,
},
@@ -16,7 +16,7 @@ use std::{
sync::{atomic::Ordering, Arc},
time::{Duration, Instant},
};
use tokio::time::sleep;
use tokio::{sync::broadcast, time::sleep};
// Functional source split intentionally keeps items in the existing module namespace.
include!("engine/runtime.rs");
@@ -30,6 +30,7 @@ include!("engine/groups.rs");
include!("engine/zone_actions.rs");
include!("engine/zone_control.rs");
include!("engine/history.rs");
include!("engine/energy.rs");
include!("engine/temperature.rs");
include!("engine/targets.rs");
include!("engine/schedules.rs");
+209 -7
View File
@@ -29,6 +29,16 @@ async fn send_command_locked_inner(
if !device.enabled {
return Err(AppError::BadRequest("device is disabled".into()));
}
if device.connection_type == ConnectionType::GreeCloud {
return send_cloud_command_locked_inner(
state,
device,
command,
dedupe_against_cache,
track_controller_command,
)
.await;
}
// Routine control avoids redundant frames. Explicit safety transitions (global/group OFF,
// detach) may bypass cache de-duplication so stale state cannot leave a unit powered.
@@ -57,7 +67,7 @@ async fn send_command_locked_inner(
state.db.save_device(&device)?;
} else {
if device.key.as_deref().unwrap_or_default().is_empty() {
match state.gree.bind(&device).await {
match state.providers.local().bind(&device).await {
Ok(bound) => {
device.key = Some(bound.key);
device.protocol_version = bound.protocol_version;
@@ -71,14 +81,14 @@ async fn send_command_locked_inner(
}
}
}
match state.gree.command(&device, &command, suppress_beep).await {
match state.providers.local().command(&device, &command, suppress_beep).await {
Ok(result) => applied_command = result,
Err(first_err) => {
// A lost command ACK does not mean the command was lost. Read the device
// first and avoid sending the same frame (and another beep) when the requested
// state is already present. Only rebind when the verification read also fails.
let mut observed = device.clone();
let retry_result = match state.gree.poll(&mut observed).await {
let retry_result = match state.providers.local().poll(&mut observed).await {
Ok(()) if command.changed_from(&observed).is_empty() => {
device = observed;
confirmed_state = true;
@@ -91,15 +101,15 @@ async fn send_command_locked_inner(
if remaining.is_empty() {
Ok(command.clone())
} else {
state.gree.command(&device, &remaining, suppress_beep).await
state.providers.local().command(&device, &remaining, suppress_beep).await
}
}
Err(_) => match state.gree.bind(&device).await {
Err(_) => match state.providers.local().bind(&device).await {
Ok(bound) => {
device.key = Some(bound.key);
device.protocol_version = bound.protocol_version;
state.db.save_device(&device)?;
state.gree.command(&device, &command, suppress_beep).await
state.providers.local().command(&device, &command, suppress_beep).await
}
Err(_) => Err(first_err),
},
@@ -132,7 +142,7 @@ async fn send_command_locked_inner(
sleep(Duration::from_millis(delay_ms)).await;
}
let mut observed = device.clone();
match state.gree.poll(&mut observed).await {
match state.providers.local().poll(&mut observed).await {
Ok(()) => {
let requested_matches = applied_command.changed_from(&observed).is_empty();
device = observed;
@@ -231,6 +241,198 @@ async fn send_command_locked_inner(
Ok(device)
}
async fn send_cloud_command_locked_inner(
state: &AppState,
mut device: Device,
command: DeviceCommand,
dedupe_against_cache: bool,
track_controller_command: bool,
) -> Result<Device, AppError> {
validate_cloud_command_capabilities(&device, &command)?;
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 baseline = device.clone();
let suppress_beep = state.settings.read().await.suppress_device_beep;
let cloud_settings = state.settings.read().await.gree_cloud.clone();
let all_devices = state.db.list_devices()?;
let response_started = Instant::now();
// Optimistic UI state is explicit and reversible. The provider never falls back to UDP.
device.pending_command = true;
command.apply(&mut device);
if let Some(target) = command.target_temperature {
let step = device.capabilities.temperature_step.max(0.5);
let rounded = (target / step).round() * step;
device.target_temperature = rounded.clamp(
device.capabilities.min_temperature,
device.capabilities.max_temperature,
);
}
device.updated_at = Utc::now();
state.db.save_device(&device)?;
state.broadcast("device.updated", serde_json::to_value(&device).unwrap_or_default());
let mut transport_device = baseline.clone();
let applied = match state
.providers
.cloud()
.command(
&cloud_settings,
&all_devices,
&mut transport_device,
&command,
suppress_beep,
)
.await
{
Ok(applied) => applied,
Err(err) => {
let mut restored = baseline.clone();
restored.pending_command = false;
register_cloud_failure(&mut restored, &err.to_string());
state.db.save_device(&restored)?;
state.broadcast("device.updated", serde_json::to_value(&restored).unwrap_or_default());
state.log(
"warn",
"gree_cloud.command_rejected",
&format!("Cloud command failed for {}", restored.name),
json!({"device_id": restored.id, "error": cloud_public_error(&err.to_string())}),
);
return Err(AppError::Dependency(cloud_public_error(&err.to_string())));
}
};
// The reference cloud client treats a missing 2s command ACK as uncertain success.
// Do not turn that into a synchronous chain of 10s status reads: it makes Manual Control
// look broken even when the unit accepted the command. Publish the optimistic state now;
// MQTT push is the primary confirmation path and one delayed recovery read is scheduled.
let mut accepted = transport_device;
accepted.pending_command = true;
accepted.response_time_ms = Some(
response_started
.elapsed()
.as_millis()
.min(u64::MAX as u128) as u64,
);
accepted.updated_at = Utc::now();
accepted.refresh_capabilities();
state.db.save_device(&accepted)?;
if track_controller_command && !command_manual_control_fields(&applied).is_empty() {
remember_controller_command(state, &accepted.id, &applied, &baseline).await;
}
record_device_transition_timestamps(state, &baseline, &accepted)?;
state.log(
"info",
"gree_cloud.command_sent",
&format!("Updated {} through GREE Cloud", accepted.name),
json!({
"device_id": accepted.id,
"command": applied,
"confirmation": "mqtt_push_or_recovery_poll",
}),
);
state.broadcast(
"device.updated",
serde_json::to_value(&accepted).unwrap_or_default(),
);
// Recovery is deliberately asynchronous. It waits until the command handler releases
// the per-device lock, then performs one normal provider poll. A successful push may
// already have confirmed the state by then; the poll is only a bounded fallback.
let recovery_state = state.clone();
let recovery_device_id = accepted.id.clone();
tokio::spawn(async move {
sleep(Duration::from_millis(1500)).await;
let Some(_cloud_poll_guard) = recovery_state.try_begin_cloud_poll(&recovery_device_id) else {
tracing::debug!(device=%recovery_device_id, "skipping duplicate GREE Cloud recovery poll");
return;
};
if let Err(err) = poll_one(&recovery_state, &recovery_device_id).await {
tracing::debug!(
device=%recovery_device_id,
error=?err,
"GREE Cloud post-command recovery poll did not confirm state"
);
}
});
Ok(accepted)
}
fn validate_cloud_command_capabilities(
device: &Device,
command: &DeviceCommand,
) -> Result<(), AppError> {
let unsupported = if command.swing_vertical.is_some() && !device.capabilities.vertical_swing {
Some("vertical swing")
} else if command.swing_horizontal.is_some() && !device.capabilities.horizontal_swing {
Some("horizontal swing")
} else if command.quiet.is_some() && device.supports_quiet == Some(false) {
Some("quiet")
} else if command.turbo.is_some() && device.supports_turbo == Some(false) {
Some("turbo")
} else if command.light.is_some() && device.supports_light == Some(false) {
Some("light")
} else if command.air.is_some() && device.supports_air == Some(false) {
Some("air")
} else if command.xfan.is_some() && device.supports_xfan == Some(false) {
Some("X-Fan")
} else if command.health.is_some() && device.supports_health == Some(false) {
Some("health")
} else if command.sleep.is_some() && device.supports_sleep == Some(false) {
Some("sleep")
} else {
None
};
if let Some(feature) = unsupported {
return Err(AppError::BadRequest(format!(
"{} does not report support for {feature}",
device.name
)));
}
Ok(())
}
fn register_cloud_failure(device: &mut Device, error: &str) {
device.communication_failures = device.communication_failures.saturating_add(1);
device.online = false;
device.connection_status = cloud_connection_status(error);
device.response_time_ms = None;
if device.last_seen.is_none() {
device.last_cloud_sync = None;
}
device.last_error = Some(cloud_public_error(error));
device.updated_at = Utc::now();
}
fn cloud_connection_status(error: &str) -> ConnectionStatus {
let error = error.to_ascii_lowercase();
if error.contains("authentication") || error.contains("not authorized") || error.contains("invalid username") {
ConnectionStatus::AuthenticationError
} else if error.contains("mqtt") || error.contains("connect") || error.contains("tls") || error.contains("network") {
ConnectionStatus::CloudDisconnected
} else {
ConnectionStatus::Offline
}
}
fn cloud_public_error(error: &str) -> String {
let lower = error.to_ascii_lowercase();
if lower.contains("password") || lower.contains("token") || lower.contains("authorization") {
"GREE Cloud authentication failed".into()
} else {
error.chars().take(300).collect()
}
}
fn record_device_transition_timestamps(
state: &AppState,
before: &Device,
+147
View File
@@ -0,0 +1,147 @@
const ENERGY_ANOMALY_MAX_DELTA_KWH: f64 = 100.0;
fn cumulative_energy_delta(previous_kwh: Option<f64>, current_kwh: f64) -> (f64, &'static str, bool) {
let Some(previous_kwh) = previous_kwh else {
return (0.0, "baseline", false);
};
let delta = current_kwh - previous_kwh;
if delta.abs() < 0.000_000_1 {
(0.0, "duplicate", false)
} else if delta < 0.0 {
(0.0, "reset", true)
} else if delta > ENERGY_ANOMALY_MAX_DELTA_KWH {
(0.0, "anomaly_large_jump", true)
} else {
(delta.max(0.0), "ok", false)
}
}
fn normalize_energy_kwh(raw_value: f64, unit: &str) -> Result<f64, AppError> {
if !raw_value.is_finite() || raw_value < 0.0 {
return Err(AppError::BadRequest("energy meter value must be a finite non-negative number".into()));
}
match unit.trim().to_ascii_lowercase().as_str() {
"kwh" => Ok(raw_value),
"wh" => Ok(raw_value / 1000.0),
"0.1kwh" | "0.1 kwh" => Ok(raw_value * 0.1),
other => Err(AppError::BadRequest(format!("unsupported energy unit: {other}"))),
}
}
pub(crate) fn record_cumulative_energy_sample(
state: &AppState,
device_id: &str,
source: &str,
raw_value: f64,
raw_unit: &str,
current_power_kw: Option<f64>,
) -> Result<EnergyReading, AppError> {
let normalized = normalize_energy_kwh(raw_value, raw_unit)?;
let previous = state.db.last_energy_reading(device_id, source)?;
let (consumption, quality, reset_detected) = cumulative_energy_delta(
previous.as_ref().map(|row| row.normalized_meter_kwh),
normalized,
);
let reading = EnergyReading {
id: 0,
device_id: device_id.to_string(),
timestamp: Utc::now(),
source: source.to_string(),
raw_meter_value: raw_value,
raw_unit: raw_unit.to_string(),
normalized_meter_kwh: normalized,
consumption_kwh: consumption.max(0.0),
current_power_kw: current_power_kw.filter(|value| value.is_finite() && *value >= 0.0),
quality: quality.to_string(),
reset_detected,
};
state.db.add_energy_reading(&reading)?;
queue_influx_energy(state, reading.clone());
Ok(reading)
}
fn queue_influx_energy(state: &AppState, reading: EnergyReading) {
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_energy(&state.http, &settings, &reading).await {
tracing::warn!(error=?err, device_id=%reading.device_id, source=%reading.source, "cannot write energy metric to InfluxDB");
}
});
}
async fn sample_home_assistant_energy_device(state: &AppState, device: &Device) -> Result<(), AppError> {
let Some(entity_id) = device.ha_energy_entity_id.as_deref().filter(|value| !value.trim().is_empty()) else {
return Ok(());
};
let settings = state.settings.read().await.home_assistant.clone();
let payload = home_assistant::read_entity(&state.http, &settings, Some(entity_id))
.await
.map_err(|err| AppError::Device(err.to_string()))?;
let state_value = payload.get("state").and_then(Value::as_str).unwrap_or_default();
if matches!(state_value, "" | "unknown" | "unavailable") { return Ok(()); }
let raw_value: f64 = state_value.parse().map_err(|_| AppError::Device("Home Assistant energy state is not numeric".into()))?;
let attrs = payload.get("attributes").and_then(Value::as_object).cloned().unwrap_or_default();
let device_class = attrs.get("device_class").and_then(Value::as_str).unwrap_or_default();
let state_class = attrs.get("state_class").and_then(Value::as_str).unwrap_or_default();
let unit = attrs.get("unit_of_measurement").and_then(Value::as_str).unwrap_or_default();
if device_class != "energy" || !matches!(state_class, "total" | "total_increasing") {
return Err(AppError::BadRequest("selected Home Assistant entity is not a cumulative energy sensor".into()));
}
if !matches!(unit.to_ascii_lowercase().as_str(), "wh" | "kwh") {
return Err(AppError::BadRequest("Home Assistant energy sensor must use Wh or kWh".into()));
}
let _ = record_cumulative_energy_sample(state, &device.id, "home_assistant", raw_value, unit, None)?;
Ok(())
}
pub(crate) async fn home_assistant_energy_loop(state: AppState) {
sleep(Duration::from_secs(10)).await;
loop {
let settings = state.settings.read().await.home_assistant.clone();
if !settings.url.trim().is_empty() && !settings.token.trim().is_empty() {
if let Ok(devices) = state.db.list_devices() {
for device in devices.into_iter().filter(|device| device.enabled && device.ha_energy_entity_id.is_some()) {
if let Err(err) = sample_home_assistant_energy_device(&state, &device).await {
tracing::warn!(device=%device.id, error=?err, "Home Assistant energy sample failed");
}
}
}
}
sleep(Duration::from_secs(60)).await;
}
}
#[cfg(test)]
mod energy_tests {
use super::*;
#[test]
fn normalizes_wh_and_kwh() {
assert!((normalize_energy_kwh(1500.0, "Wh").unwrap() - 1.5).abs() < 1e-9);
assert!((normalize_energy_kwh(1.5, "kWh").unwrap() - 1.5).abs() < 1e-9);
}
#[test]
fn cumulative_counter_becomes_non_negative_delta() {
assert_eq!(cumulative_energy_delta(None, 152.1), (0.0, "baseline", false));
let (delta, quality, reset) = cumulative_energy_delta(Some(152.1), 152.4);
assert!((delta - 0.3).abs() < 1e-9);
assert_eq!(quality, "ok");
assert!(!reset);
assert_eq!(cumulative_energy_delta(Some(152.4), 152.4), (0.0, "duplicate", false));
assert_eq!(cumulative_energy_delta(Some(153.0), 1.0), (0.0, "reset", true));
assert_eq!(cumulative_energy_delta(Some(1.0), 150.0), (0.0, "anomaly_large_jump", true));
}
#[test]
fn unit_changes_normalize_before_delta() {
let wh = normalize_energy_kwh(152_400.0, "Wh").unwrap();
let kwh = normalize_energy_kwh(153.0, "kWh").unwrap();
let (delta, quality, reset) = cumulative_energy_delta(Some(wh), kwh);
assert!((delta - 0.6).abs() < 1e-9);
assert_eq!(quality, "ok");
assert!(!reset);
}
}
+2 -1
View File
@@ -141,7 +141,8 @@ fn manual_override_matches_baseline(zone: &Zone, device: &Device) -> bool {
"power" => device.power == baseline.power,
"mode" => device.mode == baseline.mode,
"target_temperature" => {
device.target_temperature.round() == baseline.target_temperature.round()
let step = device.capabilities.temperature_step.max(0.5);
(device.target_temperature / step).round() == (baseline.target_temperature / step).round()
}
"fan_speed" => device.fan_speed == baseline.fan_speed,
"quiet" => device.quiet == baseline.quiet,
+204 -15
View File
@@ -56,14 +56,58 @@ async fn poll_one_locked(state: &AppState, device_id: &str) -> Result<Device, Ap
}
pub(crate) async fn poll_all(state: &AppState) -> Result<()> {
let device_ids: Vec<String> = state.db.list_devices()?.into_iter()
.filter(|device| device.enabled)
.map(|device| device.id)
.collect();
for device_id in device_ids {
let _zone_guards = lock_poll_zone_operations(state, &device_id).await?;
let _device_guard = state.lock_device_operation(&device_id).await;
let _ = poll_one_locked(state, &device_id).await?;
let devices = state.db.list_devices()?;
let cloud_interval = state
.settings
.read()
.await
.gree_cloud
.polling_interval_seconds
.max(30);
let startup = !state.initial_device_sync_complete.load(Ordering::Acquire);
// LAN always runs first and keeps its historical sequential locking/transport behavior.
// Cloud work is detached afterwards so an Internet/broker timeout cannot delay UDP cycles.
for device in devices.iter().filter(|device| device.enabled && device.connection_type == ConnectionType::Local) {
let _zone_guards = lock_poll_zone_operations(state, &device.id).await?;
let _device_guard = state.lock_device_operation(&device.id).await;
let _ = poll_one_locked(state, &device.id).await?;
}
for device in devices.into_iter().filter(|device| device.enabled && device.connection_type == ConnectionType::GreeCloud) {
// A failed Cloud read has no last_cloud_sync, so use updated_at (which is refreshed
// on failures) as the retry baseline. Otherwise an offline unit would be considered
// due on every fast LAN poll cycle and detached tasks would accumulate indefinitely.
let retry_baseline = device.last_cloud_sync.unwrap_or(device.updated_at);
let due = startup
|| Utc::now()
.signed_duration_since(retry_baseline)
.num_seconds()
>= cloud_interval as i64;
if !due { continue; }
if startup {
// Persist a conservative startup state before the asynchronous Cloud read. This
// prevents a stale pre-restart Online snapshot from driving thermostat commands.
let mut pending = device.clone();
pending.online = false;
pending.connection_status = ConnectionStatus::Unknown;
pending.response_time_ms = None;
state.db.save_device(&pending)?;
}
// Do not enqueue another detached poll while one for this Cloud device is still
// running or waiting on its operation lock. This is deliberately outside Tokio's
// async lock graph so an offline device cannot create an ever-growing waiter queue.
let Some(cloud_poll_guard) = state.try_begin_cloud_poll(&device.id) else {
continue;
};
let state = state.clone();
let device_id = device.id.clone();
tokio::spawn(async move {
let _cloud_poll_guard = cloud_poll_guard;
if let Err(err) = poll_one(&state, &device_id).await {
tracing::warn!(device=%device_id, error=?err, "GREE Cloud fallback poll failed");
}
});
}
Ok(())
}
@@ -82,6 +126,35 @@ pub(crate) async fn poll_all_locked(state: &AppState) -> Result<()> {
}
async fn poll_device(state: &AppState, device: &mut Device) {
if device.connection_type == ConnectionType::GreeCloud {
let previous_failures = device.communication_failures;
let response_started = Instant::now();
let cloud_settings = state.settings.read().await.gree_cloud.clone();
let all_devices = match state.db.list_devices() {
Ok(items) => items,
Err(err) => {
register_cloud_poll_failure(device, &err.to_string());
return;
}
};
match state.providers.cloud().poll(&cloud_settings, &all_devices, device).await {
Ok(()) => {
device.pending_command = false;
device.response_time_ms = Some(response_started.elapsed().as_millis().min(u64::MAX as u128) as u64);
device.refresh_capabilities();
if previous_failures > 0 {
state.log("info", "gree_cloud.device_online", &format!("{} is online through GREE Cloud", device.name), json!({"device_id": device.id}));
}
}
Err(err) => {
register_cloud_poll_failure(device, &err.to_string());
if previous_failures == 0 {
state.log("warn", "gree_cloud.device_offline", &format!("{} Cloud status failed", device.name), json!({"device_id": device.id, "error": cloud_poll_public_error(&err.to_string())}));
}
}
}
return;
}
if device.simulated {
simulate_tick(device);
return;
@@ -89,7 +162,7 @@ async fn poll_device(state: &AppState, device: &mut Device) {
let previous_failures = device.communication_failures;
let response_started = Instant::now();
if device.key.as_deref().unwrap_or_default().is_empty() {
match state.gree.bind(device).await {
match state.providers.local().bind(device).await {
Ok(bound) => {
device.key = Some(bound.key);
device.protocol_version = bound.protocol_version;
@@ -102,18 +175,18 @@ async fn poll_device(state: &AppState, device: &mut Device) {
}
}
}
if let Err(first_err) = state.gree.poll(device).await {
if let Err(first_err) = state.providers.local().poll(device).await {
// One lost UDP response is common on Wi-Fi and must not trigger a bind storm.
// Rebind only after at least one consecutive failed poll; a successful retry clears
// the counter in GreeClient::poll.
if previous_failures == 0 {
record_poll_failure(device, &first_err.to_string());
} else {
match state.gree.bind(device).await {
match state.providers.local().bind(device).await {
Ok(bound) => {
device.key = Some(bound.key);
device.protocol_version = bound.protocol_version;
if let Err(err) = state.gree.poll(device).await {
if let Err(err) = state.providers.local().poll(device).await {
record_poll_failure(device, &err.to_string());
}
}
@@ -122,11 +195,40 @@ async fn poll_device(state: &AppState, device: &mut Device) {
}
}
if device.communication_failures == 0 && device.online {
device.connection_status = ConnectionStatus::Online;
device.response_time_ms = Some(response_started.elapsed().as_millis().min(u64::MAX as u128) as u64);
}
log_poll_health_transition(state, device, previous_failures).await;
}
fn register_cloud_poll_failure(device: &mut Device, error: &str) {
device.communication_failures = device.communication_failures.saturating_add(1);
device.online = false;
let lower = error.to_ascii_lowercase();
device.connection_status = if lower.contains("authentication") || lower.contains("not authorized") {
ConnectionStatus::AuthenticationError
} else if lower.contains("mqtt") || lower.contains("connect") || lower.contains("tls") {
ConnectionStatus::CloudDisconnected
} else {
ConnectionStatus::Offline
};
device.response_time_ms = None;
if device.last_seen.is_none() {
device.last_cloud_sync = None;
}
device.last_error = Some(cloud_poll_public_error(error));
device.updated_at = Utc::now();
}
fn cloud_poll_public_error(error: &str) -> String {
let lower = error.to_ascii_lowercase();
if lower.contains("password") || lower.contains("token") || lower.contains("authorization") {
"GREE Cloud authentication failed".into()
} else {
error.chars().take(300).collect()
}
}
async fn log_poll_health_transition(state: &AppState, device: &Device, previous_failures: u8) {
let threshold = state.settings.read().await.notifications.communication_failure_threshold.max(2);
let current_failures = u32::from(device.communication_failures);
@@ -190,7 +292,7 @@ fn record_reading(state: &AppState, device: &Device) -> Result<()> {
fn record_poll_failure(device: &mut Device, error: &str) {
device.communication_failures = device.communication_failures.saturating_add(1);
// A single dropped UDP response is not enough to declare an AC offline.
if device.communication_failures >= 3 { device.online = false; }
if device.communication_failures >= 3 { device.online = false; device.connection_status = ConnectionStatus::Offline; }
device.last_error = Some(error.to_string());
device.updated_at = Utc::now();
}
@@ -278,7 +380,11 @@ fn command_field_matches_device(command: &DeviceCommand, field: &str, device: &D
"power" => command.power.map(|value| value == device.power).unwrap_or(false),
"mode" => command.mode.as_deref().map(|value| value == device.mode.as_str()).unwrap_or(false),
"target_temperature" => command.target_temperature
.map(|value| value.clamp(8.0, 30.0).round() == device.target_temperature.clamp(8.0, 30.0).round())
.map(|value| {
let step = device.capabilities.temperature_step.max(0.5);
(value.clamp(device.capabilities.min_temperature, device.capabilities.max_temperature) / step).round()
== (device.target_temperature.clamp(device.capabilities.min_temperature, device.capabilities.max_temperature) / step).round()
})
.unwrap_or(false),
"fan_speed" => command.fan_speed.map(|value| value.min(5) == device.fan_speed).unwrap_or(false),
"quiet" => command.quiet.map(|value| value == device.quiet).unwrap_or(false),
@@ -363,7 +469,10 @@ fn externally_changed_control_fields(before: &Device, after: &Device, zone: &Zon
let mut fields = Vec::new();
if before.power != after.power { fields.push("power".to_string()); }
if before.mode != after.mode { fields.push("mode".to_string()); }
if before.target_temperature.round() != after.target_temperature.round() {
let temperature_step = after.capabilities.temperature_step.max(0.5);
if (before.target_temperature / temperature_step).round()
!= (after.target_temperature / temperature_step).round()
{
fields.push("target_temperature".to_string());
}
// Some GREE units accept the controller's standby Low fan hint and later report Auto
@@ -377,3 +486,83 @@ fn externally_changed_control_fields(before: &Device, after: &Device, zone: &Zon
}
pub(crate) async fn cloud_push_loop(state: AppState) {
let mut receiver = state.providers.cloud().subscribe_push();
loop {
match receiver.recv().await {
Ok(event) => {
let devices = match state.db.list_devices() {
Ok(items) => items,
Err(err) => {
tracing::warn!(error=?err, "cannot load devices for GREE Cloud push update");
continue;
}
};
let matching: Vec<String> = devices
.into_iter()
.filter(|device| {
if device.connection_type != ConnectionType::GreeCloud { return false; }
if let Some(id) = event.cloud_device_id.as_deref() {
return device.cloud_device_id.as_deref().is_some_and(|value| value.eq_ignore_ascii_case(id));
}
device.cloud_parent_mac.as_deref().is_some_and(|value| value.eq_ignore_ascii_case(&event.parent_mac))
})
.map(|device| device.id)
.collect();
for device_id in matching {
let _guard = state.lock_device_operation(&device_id).await;
let Ok(Some(mut device)) = state.db.get_device(&device_id) else { continue; };
let before = device.clone();
if !event.properties.is_empty() {
if let Some(raw_energy) = event.properties.get("ElcAll").and_then(|value| {
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::<f64>().ok()))
}) {
if let Err(err) = record_cumulative_energy_sample(
&state, &device.id, "gree_cloud", raw_energy, "0.1kWh", None
) {
tracing::warn!(device=%device.id, error=?err, "cannot record GREE Cloud energy sample");
}
}
crate::provider::apply_cloud_properties(&mut device, &event.properties);
device.pending_command = false;
device.last_cloud_sync = Some(Utc::now());
device.last_seen = Some(Utc::now());
device.connection_status = ConnectionStatus::Online;
device.online = true;
device.communication_failures = 0;
device.last_error = None;
if let Some(version) = event.cipher_version { device.protocol_version = version; }
device.refresh_capabilities();
} else if event.connected == Some(true) {
// A connect topic proves cloud presence but does not replace a status frame.
device.connection_status = ConnectionStatus::Online;
device.online = true;
device.last_seen = Some(Utc::now());
}
device.updated_at = Utc::now();
if let Err(err) = state.db.save_device(&device) {
tracing::warn!(device=%device_id, error=?err, "cannot persist GREE Cloud push state");
continue;
}
if !event.properties.is_empty() {
let _ = record_reading(&state, &device);
if let Err(err) = detect_external_device_control(&state, &before, &device).await {
tracing::warn!(device=%device_id, error=?err, "cannot process external GREE Cloud state change");
}
}
state.broadcast_with_control_plan_invalidation(
"device.updated",
serde_json::to_value(&device).unwrap_or_default(),
device_runtime_change_affects_control_plan(&before, &device),
);
}
}
Err(broadcast::error::RecvError::Lagged(skipped)) => tracing::warn!(skipped, "GREE Cloud push state receiver lagged"),
Err(broadcast::error::RecvError::Closed) => break,
}
}
}
+40
View File
@@ -25,6 +25,27 @@ pub fn start(state: AppState) {
tracing::warn!(error=?err, "cannot reset temporary thermostat observation continuity after restart");
}
let cloud_reconnect_provider = state.providers.cloud().clone();
let cloud_reconnect_settings = state.settings.clone();
let cloud_reconnect_db = state.db.clone();
tokio::spawn(async move {
crate::provider::cloud_reconnect_loop(
cloud_reconnect_provider,
cloud_reconnect_settings,
cloud_reconnect_db,
)
.await;
});
let cloud_push_state = state.clone();
tokio::spawn(async move {
cloud_push_loop(cloud_push_state).await;
});
let ha_energy_state = state.clone();
tokio::spawn(async move {
home_assistant_energy_loop(ha_energy_state).await;
});
let control_plan_state = state.clone();
tokio::spawn(async move {
control_plan_cache_loop(control_plan_state).await;
@@ -144,6 +165,13 @@ pub fn start(state: AppState) {
Ok(_) => {}
Err(err) => tracing::warn!(error=?err, "cannot prune readings"),
}
match maintenance_state.db.prune_energy_readings(retention_days) {
Ok(count) if count > 0 => {
tracing::info!(count, retention_days, "old energy readings pruned")
}
Ok(_) => {}
Err(err) => tracing::warn!(error=?err, "cannot prune energy readings"),
}
}
let event_retention_days = settings.event_log_retention_days.max(1) as i64;
match maintenance_state.db.prune_events(event_retention_days) {
@@ -176,5 +204,17 @@ async fn archive_old_history(state: &AppState, threshold_days: u32) -> Result<u6
break;
}
}
for _ in 0..50 {
let energy = state.db.energy_before(cutoff, 1_000)?;
if energy.is_empty() {
break;
}
influxdb::write_energy_batch(&state.http, &settings, &energy).await?;
let deleted = state.db.delete_energy_batch(&energy)?;
moved += deleted;
if deleted == 0 {
break;
}
}
Ok(moved)
}
+21 -2
View File
@@ -824,9 +824,28 @@ async fn control_zones(state: &AppState) -> Result<()> {
// frame as the standby setpoint and Low fan. When demand returns, disable Quiet
// on the normal smart-fan transition. When scheduled night mode owns Quiet, it
// explicitly enables it inside the window and releases it outside the window.
let quiet_supported = if device.connection_type == ConnectionType::GreeCloud {
device.capabilities.quiet
} else {
state
.providers
.local()
.client()
.quiet_command_supported(&device.id)
};
let sleep_supported = if device.connection_type == ConnectionType::GreeCloud {
device.capabilities.sleep
} else {
device.supports_sleep == Some(true)
&& state
.providers
.local()
.client()
.sleep_command_supported(&device.id)
};
let desired_quiet = smart_quiet_command(
zone.smart_fan,
state.gree.quiet_command_supported(&device.id),
quiet_supported,
previous_demand,
zone.demand,
device.quiet,
@@ -838,7 +857,7 @@ async fn control_zones(state: &AppState) -> Result<()> {
settings.night_mode.enabled,
night_active,
settings.night_mode.use_native_sleep,
device.supports_sleep == Some(true) && state.gree.sleep_command_supported(&device.id),
sleep_supported,
device.sleep,
);
+5
View File
@@ -18,6 +18,8 @@ pub enum AppError {
Unauthorized,
#[error("device communication failed: {0}")]
Device(String),
#[error("external dependency failed: {0}")]
Dependency(String),
#[error(transparent)]
Internal(#[from] anyhow::Error),
}
@@ -30,6 +32,9 @@ impl IntoResponse for AppError {
Self::Conflict(v) => (StatusCode::CONFLICT, v.clone()),
Self::Unauthorized => (StatusCode::UNAUTHORIZED, "unauthorized".into()),
Self::Device(v) => (StatusCode::BAD_GATEWAY, v.clone()),
// Cloud/provider outages are an expected external-dependency failure, not a
// controller-side 502. Keep Local's historical Device -> 502 mapping unchanged.
Self::Dependency(v) => (StatusCode::FAILED_DEPENDENCY, v.clone()),
Self::Internal(v) => {
tracing::error!(error = ?v, "internal error");
(
+37
View File
@@ -204,6 +204,43 @@ pub async fn read_entity(
response.json().await.context("invalid Home Assistant JSON")
}
/// Read the Home Assistant state registry. Callers must filter the result before exposing it.
pub async fn list_entities(
default_client: &reqwest::Client,
settings: &HomeAssistantSettings,
) -> Result<Vec<Value>> {
if settings.url.trim().is_empty() {
bail!("Home Assistant URL is not configured")
}
if settings.token.trim().is_empty() {
bail!("Home Assistant token is not configured")
}
let mut base = Url::parse(settings.url.trim()).context("invalid Home Assistant URL")?;
if !matches!(base.scheme(), "http" | "https") {
bail!("Home Assistant URL must use http or https")
}
base = base.join("api/states").context("cannot build Home Assistant API URL")?;
let response = request_client(default_client, settings)?
.get(base)
.bearer_auth(settings.token.trim())
.header("Accept", "application/json")
.send()
.await
.context("Home Assistant state registry request failed")?;
let status = response.status();
if !status.is_success() {
let body = response.text().await.unwrap_or_default();
bail!(
"Home Assistant returned {status}: {}",
body.chars().take(200).collect::<String>()
)
}
response
.json::<Vec<Value>>()
.await
.context("invalid Home Assistant state registry JSON")
}
/// Read the raw Home Assistant state for Flow conditions. Unlike `read_temperature`, this
/// intentionally keeps the state as text so binary_sensor, switch, input_boolean and custom
/// entities can participate in visual automations.
+2 -1
View File
@@ -4,11 +4,12 @@ use reqwest::Client;
use serde_json::Value;
use std::collections::HashMap;
use crate::models::{HaReading, InfluxDbSettings, Reading, ZoneReading};
use crate::models::{EnergyReading, HaReading, InfluxDbSettings, 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";
// Functional source split intentionally keeps items in the existing module namespace.
include!("influxdb/write.rs");
+84
View File
@@ -460,3 +460,87 @@ async fn query_v2(
}
Ok(rows)
}
pub async fn query_energy(
client: &Client,
settings: &InfluxDbSettings,
device_id: &str,
source: Option<&str>,
start: DateTime<Utc>,
stop: DateTime<Utc>,
bucket_seconds: i64,
limit: u32,
) -> Result<Vec<EnergyReading>> {
if settings.version == "1" {
let source_filter = source
.map(|value| format!(" AND \"source\"='{}'", influxql_string(value)))
.unwrap_or_default();
let q = format!(
"SELECT sum(\"consumption_kwh\") AS \"consumption_kwh\" FROM \"{ENERGY_MEASUREMENT}\" WHERE time >= '{}' AND time < '{}' AND \"device_id\"='{}'{} GROUP BY time({}s),\"device_id\",\"source\" fill(none) LIMIT {}",
start.to_rfc3339(),
stop.to_rfc3339(),
influxql_string(device_id),
source_filter,
bucket_seconds.max(1),
limit
);
let series = query_v1(client, settings, &q).await?;
let mut out = Vec::new();
for item in series {
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(EnergyReading {
id: 0,
device_id: device_id.to_string(),
timestamp,
source: source.clone(),
raw_meter_value: 0.0,
raw_unit: "kWh".into(),
normalized_meter_kwh: 0.0,
consumption_kwh: row_num(&row, "consumption_kwh").unwrap_or(0.0).max(0.0),
current_power_kw: None,
quality: "influx_aggregate".into(),
reset_detected: false,
});
}
}
out.sort_by_key(|row| row.timestamp);
out.truncate(limit as usize);
return Ok(out);
}
let source_filter = source
.map(|value| format!(" |> filter(fn: (r) => r.source == {})", flux_string(value)))
.unwrap_or_default();
let query = format!(
"from(bucket: {}) |> range(start: time(v: {}), stop: time(v: {})) |> filter(fn: (r) => r._measurement == {} and r._field == \"consumption_kwh\" and r.device_id == {}){} |> aggregateWindow(every: {}s, fn: sum, createEmpty: false) |> sort(columns:[\"_time\"])",
flux_string(&settings.bucket),
flux_string(&start.to_rfc3339()),
flux_string(&stop.to_rfc3339()),
flux_string(ENERGY_MEASUREMENT),
flux_string(device_id),
source_filter,
bucket_seconds.max(1),
);
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; };
out.push(EnergyReading {
id: 0,
device_id: device_id.to_string(),
timestamp,
source: row.get("source").cloned().unwrap_or_default(),
raw_meter_value: 0.0,
raw_unit: "kWh".into(),
normalized_meter_kwh: 0.0,
consumption_kwh: row_f64(&row, "_value").unwrap_or(0.0).max(0.0),
current_power_kw: None,
quality: "influx_aggregate".into(),
reset_detected: false,
});
}
out.sort_by_key(|row| row.timestamp);
Ok(out)
}
+46
View File
@@ -1,3 +1,49 @@
pub async fn write_energy(
client: &Client,
settings: &InfluxDbSettings,
reading: &EnergyReading,
) -> Result<()> {
if !settings.enabled { return Ok(()); }
let mut fields = Vec::new();
push_float(&mut fields, "raw_meter_value", Some(reading.raw_meter_value));
push_float(&mut fields, "normalized_meter_kwh", Some(reading.normalized_meter_kwh));
push_float(&mut fields, "consumption_kwh", Some(reading.consumption_kwh.max(0.0)));
push_float(&mut fields, "current_power_kw", reading.current_power_kw);
push_int(&mut fields, "reset_detected", reading.reset_detected as i64);
let line = line_protocol(
ENERGY_MEASUREMENT,
&[("device_id", &reading.device_id), ("source", &reading.source), ("quality", &reading.quality), ("raw_unit", &reading.raw_unit)],
fields,
reading.timestamp,
)?;
write_line(client, settings, line).await
}
pub async fn write_energy_batch(
client: &Client,
settings: &InfluxDbSettings,
readings: &[EnergyReading],
) -> 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, "raw_meter_value", Some(reading.raw_meter_value));
push_float(&mut fields, "normalized_meter_kwh", Some(reading.normalized_meter_kwh));
push_float(&mut fields, "consumption_kwh", Some(reading.consumption_kwh.max(0.0)));
push_float(&mut fields, "current_power_kw", reading.current_power_kw);
push_int(&mut fields, "reset_detected", reading.reset_detected as i64);
lines.push(line_protocol(
ENERGY_MEASUREMENT,
&[("device_id", &reading.device_id), ("source", &reading.source), ("quality", &reading.quality), ("raw_unit", &reading.raw_unit)],
fields,
reading.timestamp,
)?);
}
write_lines(client, settings, lines.join("\n")).await
}
pub fn validate(settings: &InfluxDbSettings) -> Result<()> {
if !settings.enabled {
return Ok(());
+77 -12
View File
@@ -8,6 +8,7 @@ mod influxdb;
mod models;
mod notifications;
mod protocol;
mod provider;
mod queries;
mod state;
@@ -16,8 +17,10 @@ use config::Config;
use db::Db;
use models::Device;
use protocol::GreeClient;
use provider::ProviderDispatcher;
use state::AppState;
use std::{
future::IntoFuture,
sync::{
atomic::{AtomicBool, AtomicU64},
Arc,
@@ -46,6 +49,12 @@ async fn main() -> Result<()> {
runtime_settings.discovery_broadcast = config.discovery_broadcast.clone();
}
config.apply_runtime_env_overrides(&mut runtime_settings);
if runtime_settings.gree_cloud.installation_id.trim().is_empty() {
runtime_settings.gree_cloud.installation_id = uuid::Uuid::new_v4().to_string();
}
if runtime_settings.gree_cloud.account_id.trim().is_empty() {
runtime_settings.gree_cloud.account_id = "default".into();
}
db.save_runtime_settings(&runtime_settings)?;
if config.simulate && config.auto_seed && db.count_devices()? == 0 {
@@ -61,25 +70,43 @@ async fn main() -> Result<()> {
let (events, _) = broadcast::channel(512);
let (control_plan, _) = watch::channel(None);
let debug_gree_frames = Arc::new(AtomicBool::new(runtime_settings.debug.gree_frames));
let debug_cloud_requests = Arc::new(AtomicBool::new(runtime_settings.debug.cloud_requests));
let debug_cloud_mqtt = Arc::new(AtomicBool::new(runtime_settings.debug.cloud_mqtt));
let user_agent = format!(
"GreeController/{} (+https://git.linuxiarz.pl/gru/gree-controller-ha-addon/; instance={})",
env!("CARGO_PKG_VERSION"),
runtime_settings.gree_cloud.installation_id
);
let http = reqwest::Client::builder()
.timeout(Duration::from_secs(10))
.user_agent(concat!("gree-controller/", env!("CARGO_PKG_VERSION")))
.connect_timeout(Duration::from_secs(10))
.timeout(Duration::from_secs(30))
.user_agent(user_agent)
.build()?;
let local_client = GreeClient::new(
runtime_settings.controller_id.clone(),
(!config.gree_interface.trim().is_empty())
.then(|| config.gree_interface.trim().to_string()),
Some(events.clone()),
debug_gree_frames.clone(),
);
let providers = ProviderDispatcher::new_with_debug(
local_client,
http.clone(),
Some(events.clone()),
debug_cloud_requests.clone(),
debug_cloud_mqtt.clone(),
);
let state = AppState {
db,
settings: Arc::new(RwLock::new(runtime_settings.clone())),
config: Arc::new(config.clone()),
gree: GreeClient::new(
runtime_settings.controller_id.clone(),
(!config.gree_interface.trim().is_empty())
.then(|| config.gree_interface.trim().to_string()),
Some(events.clone()),
debug_gree_frames.clone(),
),
providers,
events,
http,
outdoor_temperature: Arc::new(RwLock::new(None)),
debug_gree_frames,
debug_cloud_requests,
debug_cloud_mqtt,
initial_device_sync_complete: Arc::new(AtomicBool::new(false)),
zone_control_wakeup: Arc::new(Notify::new()),
control_plan,
@@ -98,6 +125,7 @@ async fn main() -> Result<()> {
pending_controller_commands: Arc::new(tokio::sync::Mutex::new(
std::collections::HashMap::new(),
)),
cloud_poll_inflight: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())),
started: Instant::now(),
};
@@ -122,9 +150,35 @@ async fn main() -> Result<()> {
"GREE Controller started"
);
axum::serve(listener, app)
.with_graceful_shutdown(shutdown_signal())
.await?;
// Use a shared shutdown signal so graceful HTTP draining has a hard deadline. A stuck
// Cloud request must not make SIGTERM/Ctrl+C wait indefinitely for an open handler.
let (shutdown_tx, shutdown_rx) = watch::channel(false);
tokio::spawn(async move {
shutdown_signal().await;
let _ = shutdown_tx.send(true);
});
let server = axum::serve(listener, app)
.with_graceful_shutdown(wait_for_shutdown(shutdown_rx.clone()))
.into_future();
tokio::pin!(server);
let deadline_rx = shutdown_rx;
tokio::select! {
result = &mut server => result?,
_ = wait_for_shutdown(deadline_rx.clone()) => {
tracing::info!("shutdown requested; draining HTTP requests");
match tokio::time::timeout(Duration::from_secs(5), &mut server).await {
Ok(result) => result?,
Err(_) => tracing::warn!("HTTP graceful shutdown deadline reached; forcing runtime shutdown"),
}
}
}
if tokio::time::timeout(Duration::from_secs(2), state.providers.cloud().shutdown())
.await
.is_err()
{
tracing::warn!("GREE Cloud shutdown deadline reached");
}
tracing::info!("GREE Controller stopped");
Ok(())
}
@@ -138,6 +192,17 @@ fn init_tracing() {
.init();
}
async fn wait_for_shutdown(mut rx: watch::Receiver<bool>) {
if *rx.borrow() {
return;
}
while rx.changed().await.is_ok() {
if *rx.borrow() {
return;
}
}
}
async fn shutdown_signal() {
let ctrl_c = async {
signal::ctrl_c()
+6
View File
@@ -82,6 +82,12 @@ fn default_history_retention_days() -> u32 {
fn default_event_log_retention_days() -> u32 {
30
}
fn default_cloud_region() -> String {
"Europe".into()
}
fn default_cloud_poll_interval_seconds() -> u64 {
60
}
fn default_influx_version() -> String {
"2".into()
}
+184 -3
View File
@@ -1,8 +1,94 @@
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash, Default)]
#[serde(rename_all = "snake_case")]
pub enum ConnectionType {
#[default]
Local,
GreeCloud,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
#[serde(rename_all = "snake_case")]
pub enum ConnectionStatus {
Online,
Offline,
CloudDisconnected,
AuthenticationError,
#[default]
Unknown,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeviceCapabilities {
#[serde(default = "default_true")]
pub power: bool,
#[serde(default = "default_true")]
pub mode: bool,
#[serde(default = "default_fan_modes")]
pub fan_modes: Vec<u8>,
#[serde(default = "default_min_temperature")]
pub min_temperature: f64,
#[serde(default = "default_max_temperature")]
pub max_temperature: f64,
#[serde(default = "default_temperature_step")]
pub temperature_step: f64,
#[serde(default = "default_true")]
pub vertical_swing: bool,
#[serde(default = "default_true")]
pub horizontal_swing: bool,
#[serde(default)] pub turbo: bool,
#[serde(default)] pub quiet: bool,
#[serde(default)] pub sleep: bool,
#[serde(default)] pub light: bool,
#[serde(default)] pub health: bool,
#[serde(default)] pub buzzer_control: bool,
#[serde(default)] pub energy_meter: bool,
#[serde(default)] pub compressor_frequency: bool,
}
impl Default for DeviceCapabilities {
fn default() -> Self {
Self {
power: true, mode: true, fan_modes: default_fan_modes(),
min_temperature: 8.0, max_temperature: 30.0, temperature_step: 1.0,
vertical_swing: true, horizontal_swing: true, turbo: false, quiet: false,
sleep: false, light: false, health: false, buzzer_control: false,
energy_meter: false, compressor_frequency: false,
}
}
}
fn default_fan_modes() -> Vec<u8> { vec![0, 1, 2, 3, 4, 5] }
fn default_min_temperature() -> f64 { 8.0 }
fn default_max_temperature() -> f64 { 30.0 }
fn default_temperature_step() -> f64 { 1.0 }
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
#[serde(rename_all = "snake_case")]
pub enum EnergySourcePreference {
#[default]
Auto,
GreeCloud,
HomeAssistant,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Device {
pub id: String,
pub mac: String,
pub name: String,
/// Explicit transport. Missing in old JSON records => Local.
#[serde(default)]
pub connection_type: ConnectionType,
#[serde(default)]
pub connection_status: ConnectionStatus,
/// Stable identifier reported by the GREE Cloud account (currently the cloud MAC/CID).
#[serde(default)]
pub cloud_device_id: Option<String>,
#[serde(default)]
pub cloud_parent_mac: Option<String>,
#[serde(default)]
pub cloud_account_id: Option<String>,
pub ip: String,
#[serde(default = "default_port")]
pub port: u16,
@@ -12,7 +98,7 @@ pub struct Device {
pub model: String,
#[serde(default)]
pub firmware: String,
#[serde(default)]
#[serde(default, skip_serializing)]
pub key: Option<String>,
#[serde(default)]
pub cid: Option<String>,
@@ -70,6 +156,16 @@ pub struct Device {
#[serde(default)]
pub supports_sleep: Option<bool>,
#[serde(default)]
pub supports_buzzer_control: Option<bool>,
#[serde(default)]
pub supports_energy_meter: Option<bool>,
#[serde(default)]
pub total_energy_kwh: Option<f64>,
#[serde(default)]
pub compressor_frequency_hz: Option<f64>,
#[serde(default)]
pub last_cloud_sync: Option<DateTime<Utc>>,
#[serde(default)]
pub current_temperature: Option<f64>,
#[serde(default)]
pub outdoor_temperature: Option<f64>,
@@ -87,6 +183,22 @@ pub struct Device {
pub last_error: Option<String>,
#[serde(default)]
pub communication_failures: u8,
/// True while a Cloud command is awaiting confirmation. Local transport keeps its
/// historical synchronous UI behaviour and normally leaves this false.
#[serde(default)]
pub pending_command: bool,
#[serde(default)]
pub capabilities: DeviceCapabilities,
#[serde(default)]
pub energy_source: EnergySourcePreference,
#[serde(default)]
pub ha_energy_entity_id: Option<String>,
#[serde(default)]
pub ha_energy_unit: Option<String>,
#[serde(default)]
pub ha_energy_device_class: Option<String>,
#[serde(default)]
pub ha_energy_state_class: Option<String>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
@@ -98,6 +210,11 @@ impl Device {
id: "sim-salon".into(),
mac: "SIM000000001".into(),
name: "Living Room (simulator)".into(),
connection_type: ConnectionType::Local,
connection_status: ConnectionStatus::Online,
cloud_device_id: None,
cloud_parent_mac: None,
cloud_account_id: None,
ip: "127.0.0.1".into(),
port: 7000,
protocol_version: 1,
@@ -128,6 +245,11 @@ impl Device {
supports_xfan: Some(true),
supports_health: Some(true),
supports_sleep: Some(true),
supports_buzzer_control: Some(true),
supports_energy_meter: None,
total_energy_kwh: None,
compressor_frequency_hz: None,
last_cloud_sync: None,
current_temperature: Some(26.0),
outdoor_temperature: Some(30.0),
temperature_sensor_offset: Some(false),
@@ -136,10 +258,41 @@ impl Device {
last_seen: Some(now),
last_error: None,
communication_failures: 0,
pending_command: false,
capabilities: DeviceCapabilities {
turbo: true, quiet: true, sleep: true, light: true, health: true,
buzzer_control: true, ..DeviceCapabilities::default()
},
energy_source: EnergySourcePreference::Auto,
ha_energy_entity_id: None,
ha_energy_unit: None,
ha_energy_device_class: None,
ha_energy_state_class: None,
created_at: now,
updated_at: now,
}
}
pub fn refresh_capabilities(&mut self) {
self.capabilities.power = true;
self.capabilities.mode = true;
self.capabilities.fan_modes = default_fan_modes();
self.capabilities.min_temperature = 8.0;
self.capabilities.max_temperature = 30.0;
if self.connection_type == ConnectionType::Local {
self.capabilities.vertical_swing = true;
self.capabilities.horizontal_swing = true;
self.capabilities.temperature_step = 1.0;
}
self.capabilities.turbo = self.supports_turbo.unwrap_or(self.connection_type == ConnectionType::Local);
self.capabilities.quiet = self.supports_quiet.unwrap_or(self.connection_type == ConnectionType::Local);
self.capabilities.sleep = self.supports_sleep.unwrap_or(false);
self.capabilities.light = self.supports_light.unwrap_or(self.connection_type == ConnectionType::Local);
self.capabilities.health = self.supports_health.unwrap_or(false);
self.capabilities.buzzer_control = self.supports_buzzer_control.unwrap_or(false);
self.capabilities.energy_meter = self.supports_energy_meter.unwrap_or(false);
self.capabilities.compressor_frequency = self.compressor_frequency_hz.is_some();
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -150,6 +303,11 @@ pub struct DevicePatch {
pub protocol_version: Option<u8>,
pub key: Option<Option<String>>,
pub enabled: Option<bool>,
pub energy_source: Option<EnergySourcePreference>,
pub ha_energy_entity_id: Option<Option<String>>,
pub ha_energy_unit: Option<Option<String>>,
pub ha_energy_device_class: Option<Option<String>>,
pub ha_energy_state_class: Option<Option<String>>,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
@@ -196,7 +354,10 @@ impl DeviceCommand {
.filter(|value| value.as_str() != device.mode.as_str())
.cloned(),
target_temperature: self.target_temperature.filter(|value| {
value.clamp(8.0, 30.0).round() != device.target_temperature.clamp(8.0, 30.0).round()
let step = device.capabilities.temperature_step.max(0.5);
let requested = (value.clamp(device.capabilities.min_temperature, device.capabilities.max_temperature) / step).round();
let current = (device.target_temperature.clamp(device.capabilities.min_temperature, device.capabilities.max_temperature) / step).round();
requested != current
}),
fan_speed: self
.fan_speed
@@ -225,7 +386,10 @@ impl DeviceCommand {
device.mode = v.clone();
}
if let Some(v) = self.target_temperature {
device.target_temperature = v.clamp(8.0, 30.0).round();
let step = device.capabilities.temperature_step.max(0.5);
let min = device.capabilities.min_temperature;
let max = device.capabilities.max_temperature;
device.target_temperature = (v.clamp(min, max) / step).round() * step;
}
if let Some(v) = self.fan_speed {
device.fan_speed = v.min(5);
@@ -283,3 +447,20 @@ impl From<&Device> for ManualDeviceBaseline {
}
}
}
#[cfg(test)]
mod connection_type_tests {
use super::*;
#[test]
fn legacy_device_without_connection_type_defaults_to_local() {
let device = Device::simulated_default();
let mut value = serde_json::to_value(device).expect("serialize device");
value
.as_object_mut()
.expect("device object")
.remove("connection_type");
let migrated: Device = serde_json::from_value(value).expect("deserialize legacy device");
assert_eq!(migrated.connection_type, ConnectionType::Local);
}
}
+21
View File
@@ -49,3 +49,24 @@ pub struct EventLog {
pub message: String,
pub metadata: Value,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EnergyReading {
pub id: i64,
pub device_id: String,
pub timestamp: DateTime<Utc>,
pub source: String,
/// Original numeric value as reported by the source before unit normalization.
pub raw_meter_value: f64,
pub raw_unit: String,
/// Cumulative meter normalized to kWh.
pub normalized_meter_kwh: f64,
/// Consumption since the previous valid sample. Never negative.
pub consumption_kwh: f64,
#[serde(default)]
pub current_power_kw: Option<f64>,
pub quality: String,
#[serde(default)]
pub reset_detected: bool,
}
+4
View File
@@ -194,6 +194,10 @@ pub struct DebugSettings {
pub overlay_enabled: bool,
#[serde(default)]
pub gree_frames: bool,
#[serde(default)]
pub cloud_requests: bool,
#[serde(default)]
pub cloud_mqtt: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
+43
View File
@@ -1,3 +1,44 @@
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GreeCloudSettings {
#[serde(default)]
pub enabled: bool,
#[serde(default = "default_cloud_region")]
pub region: String,
#[serde(default)]
pub username: String,
#[serde(default)]
pub password: String,
#[serde(default = "default_cloud_poll_interval_seconds")]
pub polling_interval_seconds: u64,
/// Random, non-personal installation identifier used in the HTTP User-Agent.
#[serde(default)]
pub installation_id: String,
/// Reserved account key so the model can grow to multiple GREE accounts without changing devices.
#[serde(default)]
pub account_id: String,
#[serde(default)]
pub last_successful_contact: Option<DateTime<Utc>>,
/// Duration of the most recent successful REST operation against GREE Cloud.
#[serde(default)]
pub last_rest_response_time_ms: Option<u64>,
}
impl Default for GreeCloudSettings {
fn default() -> Self {
Self {
enabled: false,
region: default_cloud_region(),
username: String::new(),
password: String::new(),
polling_interval_seconds: default_cloud_poll_interval_seconds(),
installation_id: String::new(),
account_id: "default".into(),
last_successful_contact: None,
last_rest_response_time_ms: None,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RuntimeSettings {
pub controller_id: String,
@@ -39,6 +80,8 @@ pub struct RuntimeSettings {
#[serde(default)]
pub night_mode: NightModeSettings,
#[serde(default)]
pub gree_cloud: GreeCloudSettings,
#[serde(default)]
pub notifications: NotificationSettings,
pub home_assistant: HomeAssistantSettings,
}
+24
View File
@@ -15,6 +15,29 @@ pub struct GreeSettings {
pub compressor_protection_seconds: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GreeCloudSettingsUpdate {
pub enabled: bool,
pub region: String,
pub username: String,
#[serde(default)]
pub password: Option<String>,
pub polling_interval_seconds: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GreeCloudSettingsView {
pub enabled: bool,
pub region: String,
pub username: String,
pub password_configured: bool,
pub polling_interval_seconds: u64,
pub installation_id: String,
pub account_id: String,
pub last_successful_contact: Option<DateTime<Utc>>,
pub last_rest_response_time_ms: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HistorySettings {
pub retention_days: u32,
@@ -116,6 +139,7 @@ pub struct HomeAssistantSettingsView {
pub struct SettingsSnapshot {
pub application: ApplicationSettings,
pub gree: GreeSettings,
pub gree_cloud: GreeCloudSettingsView,
pub history: HistorySettings,
pub influxdb: InfluxDbSettingsView,
pub notifications: NotificationSettingsView,
+1 -1
View File
@@ -1,7 +1,7 @@
use super::crypto::{
decrypt_v1, decrypt_v2, encrypt_v1, encrypt_v2, GENERIC_GREE_V1_KEY, GENERIC_GREE_V2_KEY,
};
use crate::models::{ApiEvent, Device, DeviceCommand};
use crate::models::{ApiEvent, ConnectionStatus, ConnectionType, Device, DeviceCommand};
use anyhow::{anyhow, bail, Context, Result};
use chrono::Utc;
use serde_json::{json, Value};
+17
View File
@@ -179,6 +179,11 @@ impl GreeClient {
id: format!("gree-{}", mac.to_ascii_lowercase()),
mac,
name,
connection_type: ConnectionType::Local,
connection_status: ConnectionStatus::Unknown,
cloud_device_id: None,
cloud_parent_mac: None,
cloud_account_id: None,
ip: source.ip().to_string(),
port: if source.port() == 0 {
7000
@@ -213,6 +218,11 @@ impl GreeClient {
supports_xfan: None,
supports_health: None,
supports_sleep: None,
supports_buzzer_control: None,
supports_energy_meter: None,
total_energy_kwh: None,
compressor_frequency_hz: None,
last_cloud_sync: None,
current_temperature: None,
outdoor_temperature: None,
temperature_sensor_offset: None,
@@ -221,6 +231,13 @@ impl GreeClient {
last_seen: Some(now),
last_error: None,
communication_failures: 0,
pending_command: false,
capabilities: crate::models::DeviceCapabilities::default(),
energy_source: crate::models::EnergySourcePreference::Auto,
ha_energy_entity_id: None,
ha_energy_unit: None,
ha_energy_device_class: None,
ha_energy_state_class: None,
created_at: now,
updated_at: now,
}))
+476
View File
@@ -0,0 +1,476 @@
use super::crypto::decrypt_v1;
use aes::{
cipher::{generic_array::GenericArray, BlockEncrypt, KeyInit},
Aes128,
};
use anyhow::{anyhow, bail, Context, Result};
use base64::{engine::general_purpose::STANDARD, Engine as _};
use chrono::Utc;
use reqwest::header::{HeaderMap, HeaderValue, CONTENT_TYPE};
use serde::{Deserialize, Serialize};
use serde_json::{json, Map, Value};
use std::collections::HashMap;
const APP_ID: &str = "4920681951525131286";
const APP_HASH: &str = "0fa513124aa97781d1f3f40d61ca1a89";
const API_AES_KEY: &str = "#G$&^jgfujy6ujxt";
const GAEN1: &str = "5ac2bdf935bcca70";
pub fn region_base_url(region: &str) -> Option<&'static str> {
match region {
"Australia" => Some("https://augrih.gree.com"),
"China Mainland" => Some("https://grih.gree.com"),
"East South Asia" => Some("https://hkgrih.gree.com"),
"Europe" => Some("https://eugrih.gree.com"),
"India" => Some("https://ingrih.gree.com"),
"Latin American" => Some("https://lagrih.gree.com"),
"Middle East" => Some("https://megrih.gree.com"),
"North American" => Some("https://nagrih.gree.com"),
"Russia" => Some("https://rugrih.gree.com"),
"South American" => Some("https://sagrih.gree.com"),
_ => None,
}
}
pub fn parent_mac(mac: &str) -> String {
let compact = mac.trim().replace([':', '-'], "").to_ascii_uppercase();
if compact.len() > 12 && compact.ends_with("00") {
compact[..compact.len() - 2].to_string()
} else {
compact
}
}
#[derive(Debug, Clone)]
pub struct CloudCredentials {
pub user_id: i64,
pub token: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CloudDeviceInfo {
pub name: String,
pub mac: String,
pub key: String,
pub model: Option<String>,
pub version: Option<String>,
pub online: bool,
}
#[derive(Debug, Clone, Serialize)]
pub struct CloudDeviceView {
pub id: String,
pub name: String,
pub mac: String,
pub parent_mac: String,
pub model: Option<String>,
pub version: Option<String>,
pub online: bool,
pub already_added: bool,
}
#[derive(Clone)]
pub struct GreeCloudApi {
http: reqwest::Client,
base_url: String,
username: String,
password: String,
credentials: Option<CloudCredentials>,
}
impl GreeCloudApi {
pub fn for_region(
http: reqwest::Client,
region: &str,
username: &str,
password: &str,
) -> Result<Self> {
let base_url = region_base_url(region)
.ok_or_else(|| anyhow!("unknown GREE Cloud region: {region}"))?;
if username.trim().is_empty() {
bail!("GREE Cloud login/email is required");
}
if password.is_empty() {
bail!("GREE Cloud password is required");
}
Ok(Self {
http,
base_url: base_url.to_string(),
username: username.trim().to_string(),
password: password.to_string(),
credentials: None,
})
}
pub async fn login(&mut self) -> Result<CloudCredentials> {
let now = Utc::now();
let time = now.format("%Y-%m-%d %H:%M:%S").to_string();
let h = md5_hex(&(md5_hex(&self.password) + &self.password));
let psw = md5_hex(&(h + &time));
let payload = json!({
"psw": psw,
"t": time,
"user": self.username,
});
let data = self
.request_at("/App/UserLoginV2", payload, &["user", "psw", "t"], now)
.await
.context("GREE Cloud login request failed")?;
if data.get("r").and_then(Value::as_i64).is_some_and(|r| r != 200) {
let message = data
.get("msg")
.and_then(Value::as_str)
.unwrap_or("unknown authentication error");
bail!("GREE Cloud authentication failed: {message}");
}
let body = data
.get("data")
.and_then(Value::as_object)
.cloned()
.unwrap_or_else(|| data.as_object().cloned().unwrap_or_default());
let user_id = value_i64(body.get("uid"))
.ok_or_else(|| anyhow!("GREE Cloud login response is missing uid"))?;
let token = body
.get("token")
.and_then(Value::as_str)
.filter(|v| !v.is_empty())
.ok_or_else(|| anyhow!("GREE Cloud login response is missing token"))?
.to_string();
let credentials = CloudCredentials { user_id, token };
self.credentials = Some(credentials.clone());
Ok(credentials)
}
pub async fn get_all_devices(&self) -> Result<Vec<CloudDeviceInfo>> {
let credentials = self
.credentials
.as_ref()
.ok_or_else(|| anyhow!("GREE Cloud session is not authenticated"))?;
let homes = self.get_homes(credentials).await?;
let mut all = Vec::new();
for home_id in homes {
all.extend(self.get_devices(credentials, home_id).await?);
}
Ok(filter_duplicate_devices(all))
}
async fn get_homes(&self, credentials: &CloudCredentials) -> Result<Vec<i64>> {
let payload = json!({
"token": credentials.token,
"uid": credentials.user_id,
});
let data = self
.request_now("/App/GetHomes", payload, &["token", "uid"])
.await
.context("GREE Cloud home discovery failed")?;
let homes = data
.get("home")
.and_then(Value::as_array)
.ok_or_else(|| anyhow!("GREE Cloud homes response has no home list"))?;
Ok(homes
.iter()
.filter_map(|home| value_i64(home.get("id")))
.collect())
}
async fn get_devices(
&self,
credentials: &CloudCredentials,
home_id: i64,
) -> Result<Vec<CloudDeviceInfo>> {
let payload = json!({
"token": credentials.token,
"homeId": home_id,
"uid": credentials.user_id,
});
let data = self
.request_now(
"/App/GetDevsInRoomsOfHomeV2",
payload,
&["token", "uid", "homeId"],
)
.await
.with_context(|| format!("GREE Cloud device discovery failed for home {home_id}"))?;
let rooms = data
.get("rooms")
.and_then(Value::as_array)
.ok_or_else(|| anyhow!("GREE Cloud devices response has no rooms list"))?;
let mut devices = Vec::new();
for room in rooms {
let Some(items) = room.get("devs").and_then(Value::as_array) else {
continue;
};
for item in items {
let Some(mac) = item.get("mac").and_then(Value::as_str) else {
continue;
};
let Some(key) = item.get("key").and_then(Value::as_str) else {
continue;
};
// Keep the exact MAC casing returned by GREE for MQTT. MQTT topics are
// case-sensitive and the reference client deliberately publishes/subscribes
// with the API value unchanged. Stable application IDs are normalized later.
let cloud_mac = mac.trim().replace([':', '-'], "");
devices.push(CloudDeviceInfo {
name: item
.get("name")
.and_then(Value::as_str)
.unwrap_or("GREE Cloud")
.trim()
.to_string(),
mac: cloud_mac,
key: key.trim().to_string(),
model: optional_trimmed(item.get("model")),
version: optional_trimmed(item.get("ver")),
online: item.get("online").map(value_bool).unwrap_or(true),
});
}
}
Ok(devices)
}
async fn request_now(&self, endpoint: &str, payload: Value, hash_props: &[&str]) -> Result<Value> {
self.request_at(endpoint, payload, hash_props, Utc::now()).await
}
async fn request_at(
&self,
endpoint: &str,
payload: Value,
hash_props: &[&str],
now: chrono::DateTime<Utc>,
) -> Result<Value> {
let time = now.format("%Y-%m-%d %H:%M:%S").to_string();
let epoch = now.timestamp();
let vc = md5_hex(&format!("{APP_ID}_{APP_HASH}_{time}_{epoch}"));
let payload_object = payload
.as_object()
.ok_or_else(|| anyhow!("GREE Cloud request payload must be an object"))?;
let hash_values = hash_props
.iter()
.map(|key| python_string(payload_object.get(*key).unwrap_or(&Value::Null)))
.collect::<Vec<_>>()
.join("_");
let dat_vc = md5_hex(&format!("{APP_HASH}_{hash_values}"));
let mut body = Map::new();
body.insert(
"api".into(),
json!({"appId": APP_ID, "r": epoch, "t": time, "vc": vc}),
);
body.insert("datVc".into(), Value::String(dat_vc));
for (key, value) in payload_object {
body.insert(key.clone(), value.clone());
}
let encrypted = encrypt_cloud_api(&serde_json::to_vec(&Value::Object(body))?)?;
let mut headers = HeaderMap::new();
headers.insert(
CONTENT_TYPE,
HeaderValue::from_static("application/x-www-form-urlencoded"),
);
headers.insert("gaen1", HeaderValue::from_static(GAEN1));
headers.insert("charset", HeaderValue::from_static("utf-8"));
let response = self
.http
.post(format!("{}{}", self.base_url, endpoint))
.headers(headers)
.body(encrypted)
.send()
.await
.context("cannot connect to GREE Cloud")?;
let status = response.status();
if !status.is_success() {
bail!("GREE Cloud API returned HTTP {status}");
}
let envelope: Value = response
.json()
.await
.context("invalid GREE Cloud API envelope")?;
let encrypted_response = envelope
.get("enRes")
.and_then(Value::as_str)
.ok_or_else(|| anyhow!("GREE Cloud response has no enRes field"))?;
let decrypted = decrypt_v1(API_AES_KEY, encrypted_response)
.context("cannot decrypt GREE Cloud response")?;
serde_json::from_slice(&decrypted).context("invalid decrypted GREE Cloud JSON")
}
}
fn filter_duplicate_devices(devices: Vec<CloudDeviceInfo>) -> Vec<CloudDeviceInfo> {
let mut groups: HashMap<String, Vec<CloudDeviceInfo>> = HashMap::new();
for device in devices {
groups.entry(device.key.clone()).or_default().push(device);
}
let mut filtered = Vec::new();
for mut group in groups.into_values() {
if group.len() == 1 {
filtered.push(group.remove(0));
continue;
}
let preferred: Vec<_> = group
.iter()
.filter(|device| device.mac.len() > 12 && device.mac.ends_with("00"))
.cloned()
.collect();
if preferred.is_empty() {
filtered.extend(group);
} else {
filtered.extend(preferred);
}
}
filtered.sort_by(|a, b| a.name.to_ascii_lowercase().cmp(&b.name.to_ascii_lowercase()));
filtered
}
fn optional_trimmed(value: Option<&Value>) -> Option<String> {
value
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn value_i64(value: Option<&Value>) -> Option<i64> {
value.and_then(|value| {
value
.as_i64()
.or_else(|| value.as_u64().and_then(|v| i64::try_from(v).ok()))
.or_else(|| value.as_str().and_then(|v| v.parse::<i64>().ok()))
})
}
fn value_bool(value: &Value) -> bool {
value
.as_bool()
.or_else(|| value.as_i64().map(|v| v != 0))
.or_else(|| {
value
.as_str()
.map(|v| !matches!(v.trim().to_ascii_lowercase().as_str(), "0" | "false" | "off" | ""))
})
.unwrap_or(false)
}
fn python_string(value: &Value) -> String {
match value {
Value::String(v) => v.clone(),
Value::Bool(true) => "True".into(),
Value::Bool(false) => "False".into(),
Value::Null => "None".into(),
Value::Number(v) => v.to_string(),
other => other.to_string(),
}
}
fn encrypt_cloud_api(plaintext: &[u8]) -> Result<String> {
let key = API_AES_KEY.as_bytes();
let cipher = Aes128::new_from_slice(key).map_err(|_| anyhow!("invalid cloud AES key"))?;
let pad = 16 - (plaintext.len() % 16);
let mut data = Vec::with_capacity(plaintext.len() + pad);
data.extend_from_slice(plaintext);
data.extend(std::iter::repeat(pad as u8).take(pad));
for block in data.chunks_exact_mut(16) {
cipher.encrypt_block(GenericArray::from_mut_slice(block));
}
Ok(STANDARD.encode(data))
}
// Small self-contained MD5 implementation avoids introducing another package/lockfile dependency.
fn md5_hex(input: &str) -> String {
let digest = md5(input.as_bytes());
digest.iter().map(|byte| format!("{byte:02x}")).collect()
}
fn md5(input: &[u8]) -> [u8; 16] {
const S: [u32; 64] = [
7, 12, 17, 22, 7, 12, 17, 22, 7, 12, 17, 22, 7, 12, 17, 22, 5, 9, 14, 20, 5, 9,
14, 20, 5, 9, 14, 20, 5, 9, 14, 20, 4, 11, 16, 23, 4, 11, 16, 23, 4, 11, 16, 23, 4,
11, 16, 23, 6, 10, 15, 21, 6, 10, 15, 21, 6, 10, 15, 21, 6, 10, 15, 21,
];
const K: [u32; 64] = [
0xd76aa478, 0xe8c7b756, 0x242070db, 0xc1bdceee, 0xf57c0faf, 0x4787c62a, 0xa8304613,
0xfd469501, 0x698098d8, 0x8b44f7af, 0xffff5bb1, 0x895cd7be, 0x6b901122, 0xfd987193,
0xa679438e, 0x49b40821, 0xf61e2562, 0xc040b340, 0x265e5a51, 0xe9b6c7aa, 0xd62f105d,
0x02441453, 0xd8a1e681, 0xe7d3fbc8, 0x21e1cde6, 0xc33707d6, 0xf4d50d87, 0x455a14ed,
0xa9e3e905, 0xfcefa3f8, 0x676f02d9, 0x8d2a4c8a, 0xfffa3942, 0x8771f681, 0x6d9d6122,
0xfde5380c, 0xa4beea44, 0x4bdecfa9, 0xf6bb4b60, 0xbebfbc70, 0x289b7ec6, 0xeaa127fa,
0xd4ef3085, 0x04881d05, 0xd9d4d039, 0xe6db99e5, 0x1fa27cf8, 0xc4ac5665, 0xf4292244,
0x432aff97, 0xab9423a7, 0xfc93a039, 0x655b59c3, 0x8f0ccc92, 0xffeff47d, 0x85845dd1,
0x6fa87e4f, 0xfe2ce6e0, 0xa3014314, 0x4e0811a1, 0xf7537e82, 0xbd3af235, 0x2ad7d2bb,
0xeb86d391,
];
let bit_len = (input.len() as u64).wrapping_mul(8);
let mut message = input.to_vec();
message.push(0x80);
while message.len() % 64 != 56 {
message.push(0);
}
message.extend_from_slice(&bit_len.to_le_bytes());
let mut a0 = 0x67452301u32;
let mut b0 = 0xefcdab89u32;
let mut c0 = 0x98badcfeu32;
let mut d0 = 0x10325476u32;
for chunk in message.chunks_exact(64) {
let mut m = [0u32; 16];
for (index, word) in m.iter_mut().enumerate() {
let start = index * 4;
*word = u32::from_le_bytes([
chunk[start],
chunk[start + 1],
chunk[start + 2],
chunk[start + 3],
]);
}
let (mut a, mut b, mut c, mut d) = (a0, b0, c0, d0);
for i in 0..64 {
let (f, g) = match i {
0..=15 => ((b & c) | ((!b) & d), i),
16..=31 => ((d & b) | ((!d) & c), (5 * i + 1) % 16),
32..=47 => (b ^ c ^ d, (3 * i + 5) % 16),
_ => (c ^ (b | (!d)), (7 * i) % 16),
};
let next = a
.wrapping_add(f)
.wrapping_add(K[i])
.wrapping_add(m[g]);
a = d;
d = c;
c = b;
b = b.wrapping_add(next.rotate_left(S[i]));
}
a0 = a0.wrapping_add(a);
b0 = b0.wrapping_add(b);
c0 = c0.wrapping_add(c);
d0 = d0.wrapping_add(d);
}
let mut output = [0u8; 16];
output[0..4].copy_from_slice(&a0.to_le_bytes());
output[4..8].copy_from_slice(&b0.to_le_bytes());
output[8..12].copy_from_slice(&c0.to_le_bytes());
output[12..16].copy_from_slice(&d0.to_le_bytes());
output
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn md5_matches_standard_vectors() {
assert_eq!(md5_hex(""), "d41d8cd98f00b204e9800998ecf8427e");
assert_eq!(md5_hex("abc"), "900150983cd24fb0d6963f7d28e17f72");
assert_eq!(
md5_hex("The quick brown fox jumps over the lazy dog"),
"9e107d9d372bb6826bd81d3542a419d6"
);
}
#[test]
fn parent_mac_matches_cloud_reference() {
assert_eq!(parent_mac("AABBCCDDEEFF00"), "AABBCCDDEEFF");
assert_eq!(parent_mac("AABBCCDDEEFF"), "AABBCCDDEEFF");
}
}
+525
View File
@@ -0,0 +1,525 @@
use anyhow::{anyhow, bail, Context, Result};
use serde::{Deserialize, Serialize};
use std::{
sync::{
atomic::{AtomicBool, AtomicU16, AtomicU64, Ordering},
Arc,
},
time::{Duration, SystemTime, UNIX_EPOCH},
};
use tokio::{
io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
net::TcpStream,
sync::{broadcast, mpsc},
time::{interval, timeout},
};
use tokio_rustls::{rustls, TlsConnector};
use tokio_rustls::rustls::pki_types::ServerName;
pub const MQTT_PORT: u16 = 1984;
pub const MQTT_KEEPALIVE_SECONDS: u16 = 60;
const MQTT_QUEUE_TIMEOUT: Duration = Duration::from_secs(2);
const MQTT_WRITE_TIMEOUT: Duration = Duration::from_secs(5);
const MQTT_DISCONNECT_TIMEOUT: Duration = Duration::from_millis(750);
pub fn broker_for_region(region: &str) -> Option<&'static str> {
match region {
"Australia" => Some("mqtt-au.gree.com"),
// greeclimate 1.2.1 regional broker mapping.
"China Mainland" => Some("mqtt-cn.gree.com"),
"East South Asia" => Some("mqtt-as.gree.com"),
"Europe" => Some("mqtt-eu.gree.com"),
"India" => Some("mqtt-in.gree.com"),
"Latin American" => Some("mqtt-la.gree.com"),
"Middle East" => Some("mqtt-me.gree.com"),
"North American" => Some("mqtt-na.gree.com"),
"Russia" => Some("mqtt-ru.gree.com"),
"South American" => Some("mqtt-sa.gree.com"),
_ => None,
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MqttDeviceEnvelope {
#[serde(default)]
pub cid: String,
#[serde(default)]
pub i: i64,
#[serde(default)]
pub pack: String,
#[serde(default)]
pub t: String,
#[serde(default)]
pub tcid: String,
#[serde(default)]
pub uid: serde_json::Value,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tag: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub ts: Option<i64>,
}
#[derive(Debug, Clone)]
pub enum MqttEvent {
Message { topic: String, payload: Vec<u8> },
/// Broker-level traffic such as SUBACK/PUBACK/PINGRESP. This proves the MQTT
/// session is alive without being mistaken for a response from the HVAC unit.
Traffic { kind: &'static str },
Disconnected { reason: String },
}
#[derive(Debug)]
enum WireCommand {
Subscribe(Vec<String>),
Publish { topic: String, payload: Vec<u8> },
Raw(Vec<u8>),
Ping,
Disconnect,
}
#[derive(Clone)]
pub struct MqttConnection {
tx: mpsc::Sender<WireCommand>,
connected: Arc<AtomicBool>,
}
impl MqttConnection {
pub async fn connect(
host: &str,
port: u16,
user_id: i64,
token: &str,
connect_timeout: Duration,
events: broadcast::Sender<MqttEvent>,
) -> Result<Self> {
let tcp = timeout(connect_timeout, TcpStream::connect((host, port)))
.await
.context("GREE Cloud MQTT TCP connect timed out")?
.with_context(|| format!("cannot connect to GREE Cloud MQTT {host}:{port}"))?;
tcp.set_nodelay(true).ok();
// Do not inherit the reference library's CERT_NONE workaround: certificate and
// hostname verification are intentionally enabled here.
let mut roots = rustls::RootCertStore::empty();
roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
let tls_config = rustls::ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
let connector = TlsConnector::from(Arc::new(tls_config));
let server_name = ServerName::try_from(host.to_string())
.map_err(|_| anyhow!("invalid GREE Cloud MQTT hostname"))?;
let mut stream = timeout(connect_timeout, connector.connect(server_name, tcp))
.await
.context("GREE Cloud MQTT TLS handshake timed out")?
.context("GREE Cloud MQTT TLS handshake failed")?;
let client_id = format!("app_{:016x}", rand::random::<u64>());
let connect_packet = encode_connect(
&client_id,
&user_id.to_string(),
token,
MQTT_KEEPALIVE_SECONDS,
)?;
timeout(connect_timeout, stream.write_all(&connect_packet))
.await
.context("GREE Cloud MQTT CONNECT write timed out")??;
timeout(connect_timeout, stream.flush())
.await
.context("GREE Cloud MQTT CONNECT flush timed out")??;
let (packet_type, payload) = timeout(connect_timeout, read_packet(&mut stream))
.await
.context("GREE Cloud MQTT CONNACK timed out")??;
if packet_type >> 4 != 2 || payload.len() != 2 {
bail!("invalid GREE Cloud MQTT CONNACK");
}
if payload[1] != 0 {
let reason = match payload[1] {
1 => "unacceptable protocol version",
2 => "identifier rejected",
3 => "server unavailable",
4 => "invalid username/password",
5 => "not authorized",
_ => "unknown broker error",
};
bail!("GREE Cloud MQTT authentication/connect rejected: {reason}");
}
let (reader, writer) = tokio::io::split(stream);
let (tx, rx) = mpsc::channel(64);
let connected = Arc::new(AtomicBool::new(true));
let packet_ids = Arc::new(AtomicU16::new(1));
let last_rx_ms = Arc::new(AtomicU64::new(unix_millis()));
spawn_writer(
writer,
rx,
connected.clone(),
events.clone(),
packet_ids,
);
spawn_reader(
reader,
tx.clone(),
connected.clone(),
events.clone(),
last_rx_ms.clone(),
);
spawn_keepalive(tx.clone(), connected.clone(), events.clone(), last_rx_ms);
Ok(Self {
tx,
connected,
})
}
pub fn is_connected(&self) -> bool {
self.connected.load(Ordering::Acquire)
}
pub async fn subscribe_device(&self, parent_mac: &str) -> Result<()> {
if !self.is_connected() {
bail!("GREE Cloud MQTT is not connected");
}
let topics = [
format!("response/{parent_mac}/#"),
format!("status/{parent_mac}/#"),
format!("connect/{parent_mac}"),
];
// Match the reference client: one QoS1 SUBSCRIBE per topic. This is slightly more
// verbose than a multi-filter packet but avoids broker-specific handling differences.
for topic in topics {
timeout(
MQTT_QUEUE_TIMEOUT,
self.tx.send(WireCommand::Subscribe(vec![topic])),
)
.await
.context("GREE Cloud MQTT subscribe queue timed out")?
.map_err(|_| anyhow!("GREE Cloud MQTT writer stopped"))?;
}
Ok(())
}
pub async fn publish(&self, topic: String, payload: Vec<u8>) -> Result<()> {
if !self.is_connected() {
bail!("GREE Cloud MQTT is not connected");
}
timeout(
MQTT_QUEUE_TIMEOUT,
self.tx.send(WireCommand::Publish { topic, payload }),
)
.await
.context("GREE Cloud MQTT publish queue timed out")?
.map_err(|_| anyhow!("GREE Cloud MQTT writer stopped"))
}
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.
self.connected.store(false, Ordering::Release);
let _ = timeout(
MQTT_DISCONNECT_TIMEOUT,
self.tx.send(WireCommand::Disconnect),
)
.await;
}
}
fn spawn_keepalive(
tx: mpsc::Sender<WireCommand>,
connected: Arc<AtomicBool>,
events: broadcast::Sender<MqttEvent>,
last_rx_ms: Arc<AtomicU64>,
) {
tokio::spawn(async move {
let mut tick = interval(Duration::from_secs(30));
tick.tick().await;
while connected.load(Ordering::Acquire) {
tick.tick().await;
let idle_ms = unix_millis().saturating_sub(last_rx_ms.load(Ordering::Acquire));
if idle_ms > 90_000 {
mark_disconnected(
&connected,
&events,
"MQTT heartbeat timed out waiting for broker traffic".into(),
);
let _ = timeout(MQTT_DISCONNECT_TIMEOUT, tx.send(WireCommand::Disconnect)).await;
break;
}
match timeout(MQTT_QUEUE_TIMEOUT, tx.send(WireCommand::Ping)).await {
Ok(Ok(())) => {}
Ok(Err(_)) => {
mark_disconnected(&connected, &events, "MQTT writer stopped".into());
break;
}
Err(_) => {
mark_disconnected(&connected, &events, "MQTT ping queue timed out".into());
break;
}
}
}
});
}
fn spawn_writer<W>(
mut writer: W,
mut rx: mpsc::Receiver<WireCommand>,
connected: Arc<AtomicBool>,
events: broadcast::Sender<MqttEvent>,
packet_ids: Arc<AtomicU16>,
) where
W: AsyncWrite + Unpin + Send + 'static,
{
tokio::spawn(async move {
while let Some(command) = rx.recv().await {
let result = match command {
WireCommand::Subscribe(topics) => {
let packet_id = next_packet_id(&packet_ids);
encode_subscribe(packet_id, &topics)
}
WireCommand::Publish { topic, payload } => {
let packet_id = next_packet_id(&packet_ids);
encode_publish(packet_id, &topic, &payload)
}
WireCommand::Raw(packet) => Ok(packet),
WireCommand::Ping => Ok(vec![0xC0, 0x00]),
WireCommand::Disconnect => {
let _ = timeout(MQTT_DISCONNECT_TIMEOUT, writer.write_all(&[0xE0, 0x00])).await;
let _ = timeout(MQTT_DISCONNECT_TIMEOUT, writer.flush()).await;
connected.store(false, Ordering::Release);
break;
}
};
let packet = match result {
Ok(packet) => packet,
Err(err) => {
mark_disconnected(&connected, &events, err.to_string());
break;
}
};
match timeout(MQTT_WRITE_TIMEOUT, writer.write_all(&packet)).await {
Ok(Ok(())) => {}
Ok(Err(err)) => {
mark_disconnected(&connected, &events, format!("MQTT write failed: {err}"));
break;
}
Err(_) => {
mark_disconnected(&connected, &events, "MQTT write timed out".into());
break;
}
}
match timeout(MQTT_WRITE_TIMEOUT, writer.flush()).await {
Ok(Ok(())) => {}
Ok(Err(err)) => {
mark_disconnected(&connected, &events, format!("MQTT flush failed: {err}"));
break;
}
Err(_) => {
mark_disconnected(&connected, &events, "MQTT flush timed out".into());
break;
}
}
}
});
}
fn spawn_reader<R>(
mut reader: R,
tx: mpsc::Sender<WireCommand>,
connected: Arc<AtomicBool>,
events: broadcast::Sender<MqttEvent>,
last_rx_ms: Arc<AtomicU64>,
) where
R: AsyncRead + Unpin + Send + 'static,
{
tokio::spawn(async move {
loop {
match read_packet(&mut reader).await {
Ok((header, payload)) => {
last_rx_ms.store(unix_millis(), Ordering::Release);
match header >> 4 {
3 => {
if let Err(err) = handle_publish(header, &payload, &tx, &events).await {
tracing::warn!(error=?err, "invalid GREE Cloud MQTT PUBLISH");
}
}
9 => { let _ = events.send(MqttEvent::Traffic { kind: "SUBACK" }); }
4 => { let _ = events.send(MqttEvent::Traffic { kind: "PUBACK" }); }
13 => { let _ = events.send(MqttEvent::Traffic { kind: "PINGRESP" }); }
_ => {}
}
}
Err(err) => {
mark_disconnected(&connected, &events, format!("MQTT read failed: {err}"));
break;
}
}
}
});
}
async fn handle_publish(
header: u8,
payload: &[u8],
tx: &mpsc::Sender<WireCommand>,
events: &broadcast::Sender<MqttEvent>,
) -> Result<()> {
if payload.len() < 2 {
bail!("short MQTT publish packet");
}
let topic_len = u16::from_be_bytes([payload[0], payload[1]]) as usize;
if payload.len() < 2 + topic_len {
bail!("truncated MQTT publish topic");
}
let topic = std::str::from_utf8(&payload[2..2 + topic_len])?.to_string();
let qos = (header >> 1) & 0x03;
let mut offset = 2 + topic_len;
if qos > 0 {
if payload.len() < offset + 2 {
bail!("truncated MQTT publish packet id");
}
let packet_id = u16::from_be_bytes([payload[offset], payload[offset + 1]]);
offset += 2;
if qos == 1 {
// MQTT QoS1 requires a PUBACK for every incoming publish. Send the raw four-byte
// acknowledgement through the single writer task so frame writes never interleave.
let ack = vec![0x40, 0x02, (packet_id >> 8) as u8, packet_id as u8];
send_raw_ack(tx, ack).await?;
}
}
let body = payload[offset..].to_vec();
let _ = events.send(MqttEvent::Message { topic, payload: body });
Ok(())
}
// MQTT QoS1 delivery requires PUBACK. To keep WireCommand's public operations minimal, encode
// acknowledgement as a synthetic command handled by a reserved topic marker.
async fn send_raw_ack(tx: &mpsc::Sender<WireCommand>, ack: Vec<u8>) -> Result<()> {
timeout(MQTT_QUEUE_TIMEOUT, tx.send(WireCommand::Raw(ack)))
.await
.context("GREE Cloud MQTT ACK queue timed out")?
.map_err(|_| anyhow!("GREE Cloud MQTT writer stopped"))
}
fn unix_millis() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis()
.min(u128::from(u64::MAX)) as u64
}
fn next_packet_id(ids: &AtomicU16) -> u16 {
let id = ids.fetch_add(1, Ordering::Relaxed);
if id == 0 { 1 } else { id }
}
fn mark_disconnected(
connected: &AtomicBool,
events: &broadcast::Sender<MqttEvent>,
reason: String,
) {
if connected.swap(false, Ordering::AcqRel) {
let _ = events.send(MqttEvent::Disconnected { reason });
}
}
fn encode_connect(client_id: &str, username: &str, password: &str, keepalive: u16) -> Result<Vec<u8>> {
let mut body = Vec::new();
push_utf8(&mut body, "MQTT")?;
body.push(4); // MQTT 3.1.1
body.push(0xC2); // username + password + clean session
body.extend_from_slice(&keepalive.to_be_bytes());
push_utf8(&mut body, client_id)?;
push_utf8(&mut body, username)?;
push_utf8(&mut body, password)?;
frame(0x10, body)
}
fn encode_subscribe(packet_id: u16, topics: &[String]) -> Result<Vec<u8>> {
let mut body = Vec::new();
body.extend_from_slice(&packet_id.to_be_bytes());
for topic in topics {
push_utf8(&mut body, topic)?;
body.push(1); // requested QoS 1
}
frame(0x82, body)
}
fn encode_publish(packet_id: u16, topic: &str, payload: &[u8]) -> Result<Vec<u8>> {
let mut body = Vec::new();
push_utf8(&mut body, topic)?;
body.extend_from_slice(&packet_id.to_be_bytes());
body.extend_from_slice(payload);
frame(0x32, body) // PUBLISH QoS1
}
fn frame(header: u8, body: Vec<u8>) -> Result<Vec<u8>> {
let mut out = Vec::with_capacity(body.len() + 5);
out.push(header);
encode_remaining_length(body.len(), &mut out)?;
out.extend_from_slice(&body);
Ok(out)
}
fn push_utf8(out: &mut Vec<u8>, value: &str) -> Result<()> {
let len = u16::try_from(value.as_bytes().len()).context("MQTT string is too long")?;
out.extend_from_slice(&len.to_be_bytes());
out.extend_from_slice(value.as_bytes());
Ok(())
}
fn encode_remaining_length(mut len: usize, out: &mut Vec<u8>) -> Result<()> {
if len > 268_435_455 {
bail!("MQTT packet is too large");
}
loop {
let mut digit = (len % 128) as u8;
len /= 128;
if len > 0 {
digit |= 0x80;
}
out.push(digit);
if len == 0 {
break;
}
}
Ok(())
}
async fn read_packet<R: AsyncRead + Unpin>(reader: &mut R) -> Result<(u8, Vec<u8>)> {
let header = reader.read_u8().await?;
let mut multiplier = 1usize;
let mut remaining = 0usize;
for _ in 0..4 {
let digit = reader.read_u8().await?;
remaining = remaining
.checked_add(((digit & 0x7f) as usize).saturating_mul(multiplier))
.ok_or_else(|| anyhow!("invalid MQTT remaining length"))?;
if digit & 0x80 == 0 {
let mut payload = vec![0_u8; remaining];
reader.read_exact(&mut payload).await?;
return Ok((header, payload));
}
multiplier = multiplier.saturating_mul(128);
}
bail!("invalid MQTT remaining length encoding")
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn mqtt_connect_uses_v311_and_credentials() {
let packet = encode_connect("app_123", "42", "secret", 60).unwrap();
assert_eq!(packet[0], 0x10);
assert!(packet.windows(6).any(|w| w == b"\0\x04MQTT"));
assert!(packet.windows(3).any(|w| w == b"\0\x0242"));
}
#[test]
fn region_brokers_match_cloud_reference() {
assert_eq!(broker_for_region("Europe"), Some("mqtt-eu.gree.com"));
assert_eq!(broker_for_region("North American"), Some("mqtt-na.gree.com"));
assert_eq!(broker_for_region("China Mainland"), Some("mqtt-cn.gree.com"));
}
}
+2
View File
@@ -1,4 +1,6 @@
pub mod crypto;
pub mod gree;
pub mod gree_cloud;
pub mod gree_cloud_mqtt;
pub use gree::{merge_discovered, GreeClient};
+1875
View File
File diff suppressed because it is too large Load Diff
+2
View File
@@ -16,6 +16,8 @@ pub const LIST_DEVICES: &str = "SELECT payload FROM devices ORDER BY name COLLAT
pub const GET_DEVICE_BY_ID: &str = "SELECT payload FROM devices WHERE id=?1";
pub const GET_DEVICE_BY_MAC: &str = "SELECT payload FROM devices WHERE lower(mac)=lower(?1)";
pub const DELETE_DEVICE_READINGS: &str = "DELETE FROM readings WHERE device_id=?1";
pub const DELETE_DEVICE_ENERGY_READINGS: &str =
"DELETE FROM energy_readings WHERE device_id=?1";
pub const DELETE_SCHEDULES_BY_DEVICE_ID: &str =
"DELETE FROM schedules WHERE zone_id IN (SELECT id FROM zones WHERE json_extract(payload, '$.device_id')=?1)";
pub const DELETE_ZONES_BY_DEVICE_ID: &str =
+20
View File
@@ -105,6 +105,24 @@ CREATE INDEX IF NOT EXISTS ha_readings_entity_time_idx
CREATE INDEX IF NOT EXISTS ha_readings_zone_time_idx
ON ha_readings(zone_id, timestamp DESC);
CREATE TABLE IF NOT EXISTS energy_readings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
device_id TEXT NOT NULL,
timestamp TEXT NOT NULL,
source TEXT NOT NULL,
raw_meter_value REAL NOT NULL,
raw_unit TEXT NOT NULL,
normalized_meter_kwh REAL NOT NULL,
consumption_kwh REAL NOT NULL CHECK(consumption_kwh >= 0),
current_power_kw REAL,
quality TEXT NOT NULL,
reset_detected INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS energy_readings_device_time_idx
ON energy_readings(device_id, timestamp DESC);
CREATE INDEX IF NOT EXISTS energy_readings_source_time_idx
ON energy_readings(device_id, source, timestamp DESC);
CREATE TABLE IF NOT EXISTS event_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
@@ -135,5 +153,7 @@ INSERT OR IGNORE INTO schema_migrations(version, applied_at)
VALUES (5, strftime('%Y-%m-%dT%H:%M:%fZ','now'));
INSERT OR IGNORE INTO schema_migrations(version, applied_at)
VALUES (6, strftime('%Y-%m-%dT%H:%M:%fZ','now'));
INSERT OR IGNORE INTO schema_migrations(version, applied_at)
VALUES (7, strftime('%Y-%m-%dT%H:%M:%fZ','now'));
"#;
+37 -3
View File
@@ -2,12 +2,12 @@ use crate::{
config::Config,
db::Db,
models::{ApiEvent, ControlPlan, DeviceCommand, RuntimeSettings},
protocol::GreeClient,
provider::ProviderDispatcher,
};
use chrono::Utc;
use serde_json::Value;
use std::{
collections::HashMap,
collections::{HashMap, HashSet},
sync::{
atomic::{AtomicBool, AtomicU64, Ordering},
Arc,
@@ -33,16 +33,33 @@ pub struct ControlPlanSnapshot {
pub plan: Arc<ControlPlan>,
}
pub(crate) struct CloudPollGuard {
inflight: Arc<std::sync::Mutex<HashSet<String>>>,
device_id: String,
}
impl Drop for CloudPollGuard {
fn drop(&mut self) {
let mut inflight = self
.inflight
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
inflight.remove(&self.device_id);
}
}
#[derive(Clone)]
pub struct AppState {
pub db: Db,
pub settings: Arc<RwLock<RuntimeSettings>>,
pub config: Arc<Config>,
pub gree: GreeClient,
pub providers: ProviderDispatcher,
pub events: broadcast::Sender<ApiEvent>,
pub http: reqwest::Client,
pub outdoor_temperature: Arc<RwLock<Option<f64>>>,
pub debug_gree_frames: Arc<AtomicBool>,
pub debug_cloud_requests: Arc<AtomicBool>,
pub debug_cloud_mqtt: Arc<AtomicBool>,
/// Thermostat/automation control stays passive until every enabled device has had
/// one startup poll, preventing stale persisted device state from causing restart commands.
pub initial_device_sync_complete: Arc<AtomicBool>,
@@ -68,10 +85,27 @@ pub struct AppState {
/// Short-lived expected climate state from controller-originated commands. It prevents
/// a delayed GREE status update from being mistaken for remote/manual takeover.
pub(crate) pending_controller_commands: Arc<Mutex<HashMap<String, PendingControllerCommand>>>,
/// Deduplicates detached Cloud fallback polls. Offline devices must never build an
/// unbounded queue of tasks waiting for the same per-device lock.
pub(crate) cloud_poll_inflight: Arc<std::sync::Mutex<HashSet<String>>>,
pub started: Instant,
}
impl AppState {
pub(crate) fn try_begin_cloud_poll(&self, device_id: &str) -> Option<CloudPollGuard> {
let mut inflight = self
.cloud_poll_inflight
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if !inflight.insert(device_id.to_string()) {
return None;
}
Some(CloudPollGuard {
inflight: self.cloud_poll_inflight.clone(),
device_id: device_id.to_string(),
})
}
pub async fn lock_device_operation(&self, device_id: &str) -> OwnedMutexGuard<()> {
let lock = {
let mut locks = self.device_operation_locks.lock().await;