933 lines
53 KiB
Rust
933 lines
53 KiB
Rust
|
|
pub fn automation_action_conflicts_with_thermostat(command: &DeviceCommand) -> bool {
|
|
// Power/auto-heat-cool/target are translated by apply_automatic_device_action into durable
|
|
// zone state, so they do not fight the thermostat. Dry/fan HVAC modes do not belong to the
|
|
// thermostat domain. Fan/quiet/sleep are outputs continuously owned by the regulator and
|
|
// have no independent zone override model, so a one-shot automation would be overwritten.
|
|
command.mode.as_deref().is_some_and(|mode| !matches!(mode, "auto" | "heat" | "cool"))
|
|
|| command.fan_speed.is_some() || command.quiet.is_some() || command.sleep.is_some()
|
|
}
|
|
|
|
fn device_has_thermostat_zone(device_id: &str, zones: &[Zone]) -> bool {
|
|
zones.iter().any(|zone| zone.device_id == device_id)
|
|
}
|
|
|
|
async fn run_automations(state: &AppState) -> Result<()> {
|
|
let devices = state.db.list_devices()?;
|
|
let mut automations = state.db.list_automations()?;
|
|
// Stable arbitration for same-cycle conflicts: the oldest configured rule wins, then ID.
|
|
// This avoids database row order deciding the physical outcome (M2).
|
|
automations.sort_by(|a, b| a.created_at.cmp(&b.created_at).then_with(|| a.id.cmp(&b.id)));
|
|
let mut claimed_devices = std::collections::HashSet::<String>::new();
|
|
|
|
for mut item in automations {
|
|
if !item.enabled { continue; }
|
|
let ready = automation_ready(&item);
|
|
let should_fire = match item.trigger_kind.as_str() {
|
|
// Stateful Flow blocks must keep observing while an action is in cooldown. Otherwise
|
|
// stable/change-duration state could silently span an unobserved interval.
|
|
"flow" => {
|
|
let before_runtime = item.flow_runtime.clone();
|
|
let result = flow_conditions_match(state, &devices, &mut item).await;
|
|
if item.flow_runtime != before_runtime { state.db.save_automation(&item)?; }
|
|
match result {
|
|
Ok(value) => value,
|
|
Err(err) => {
|
|
state.log("warn", "flow.condition_error", &format!("Flow automation {} condition evaluation failed: {}", item.name, err), json!({"automation_id": item.id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id}));
|
|
false
|
|
}
|
|
}
|
|
},
|
|
"temperature_above" if ready => find_temperature(&devices, item.trigger_device_id.as_deref())
|
|
.zip(item.threshold).map(|(t, threshold)| t > threshold).unwrap_or(false),
|
|
"temperature_below" if ready => find_temperature(&devices, item.trigger_device_id.as_deref())
|
|
.zip(item.threshold).map(|(t, threshold)| t < threshold).unwrap_or(false),
|
|
"time" if ready => time_automation_due(&item, Local::now()),
|
|
_ => false,
|
|
};
|
|
if !ready || !should_fire { continue; }
|
|
|
|
// API edits/deletes and execution share one short ownership window. If the rule
|
|
// changed since this cycle snapshot was taken, skip it now and evaluate the new
|
|
// definition on the next cycle instead of firing stale configuration.
|
|
let _automation_guard = state.lock_automation_operation().await;
|
|
let Some(latest_item) = state.db.get_automation(&item.id)? else { continue; };
|
|
if latest_item.updated_at != item.updated_at { continue; }
|
|
item = latest_item;
|
|
// Group membership and zone ownership may have changed after the cycle snapshot but
|
|
// before we acquired the automation lock. Reload them inside this serialized window so
|
|
// same-cycle conflict arbitration claims the actual current target set.
|
|
let zones = state.db.list_zones()?;
|
|
let groups = state.db.list_groups()?;
|
|
|
|
if item.action_group_id.is_none() && item.action_zone_id.is_none()
|
|
&& device_blocked_by_disabled_zone(&item.action_device_id, &zones)
|
|
&& item.action.power != Some(true)
|
|
{
|
|
state.log("info", "automation.blocked_by_zone", &format!("Automation {} suppressed by disabled zone", item.name), json!({
|
|
"automation_id": item.id, "device_id": item.action_device_id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id
|
|
}));
|
|
continue;
|
|
}
|
|
if item.action_group_id.is_none() && item.action_zone_id.is_none() && device_blocked_by_manual_override(&item.action_device_id, &zones) {
|
|
state.log("info", "automation.blocked_by_manual_override", &format!("Automation {} suppressed by manual device control", item.name), json!({
|
|
"automation_id": item.id, "device_id": item.action_device_id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id
|
|
}));
|
|
continue;
|
|
}
|
|
if item.action_group_id.is_none() && item.action_zone_id.is_none() && device_blocked_by_local_thermostat(&item.action_device_id, &zones) {
|
|
state.log("info", "automation.blocked_by_local_thermostat", &format!("Automation {} suppressed by local thermostat ownership", item.name), json!({
|
|
"automation_id": item.id, "device_id": item.action_device_id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id
|
|
}));
|
|
continue;
|
|
}
|
|
if item.action_group_id.is_none() && item.action_zone_id.is_none() && device_blocked_by_temporary_thermostat(&item.action_device_id, &zones) {
|
|
state.log("info", "automation.blocked_by_temporary_thermostat", &format!("Automation {} suppressed by Temporary Quick Thermostat ownership", item.name), json!({
|
|
"automation_id": item.id, "device_id": item.action_device_id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id
|
|
}));
|
|
continue;
|
|
}
|
|
if item.action_group_id.is_none() && item.action_zone_id.is_none()
|
|
&& device_has_thermostat_zone(&item.action_device_id, &zones)
|
|
&& automation_action_conflicts_with_thermostat(&item.action)
|
|
{
|
|
// Fan/quiet/sleep are outputs continuously managed by the thermostat. Unlike
|
|
// power/mode/target they cannot be translated into durable zone state, so a direct
|
|
// automation would be immediately overwritten by the next thermostat cycle.
|
|
state.log("warn", "automation.blocked_by_thermostat_owner", &format!("Automation {} suppressed because the device is owned by an enabled thermostat zone", item.name), json!({
|
|
"automation_id": item.id, "device_id": item.action_device_id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id
|
|
}));
|
|
continue;
|
|
}
|
|
|
|
let target_devices: Vec<String> = if item.action_ha_domain.is_some() {
|
|
item.action_ha_entity_id.as_ref().map(|entity_id| vec![format!("ha:{entity_id}")]).unwrap_or_default()
|
|
} else if let Some(zone_id) = item.action_zone_id.as_deref() {
|
|
zones.iter().find(|zone| zone.id == zone_id).map(|zone| vec![zone.device_id.clone()]).unwrap_or_default()
|
|
} else if let Some(group_id) = item.action_group_id.as_deref() {
|
|
groups.iter().find(|group| group.id == group_id)
|
|
.map(|group| group.zone_ids.iter()
|
|
.filter_map(|zone_id| zones.iter().find(|zone| &zone.id == zone_id).map(|zone| zone.device_id.clone()))
|
|
.collect())
|
|
.unwrap_or_default()
|
|
} else {
|
|
vec![item.action_device_id.clone()]
|
|
};
|
|
if target_devices.iter().any(|device_id| claimed_devices.contains(device_id)) {
|
|
state.log("warn", "automation.conflict", &format!("Automation {} skipped because an older due automation already claimed the same target", item.name), json!({
|
|
"automation_id": item.id,
|
|
"flow_id": item.flow_id,
|
|
"flow_node_id": item.flow_node_id,
|
|
"group_id": item.action_group_id,
|
|
"device_id": item.action_device_id,
|
|
"target_devices": target_devices,
|
|
}));
|
|
continue;
|
|
}
|
|
|
|
let result: Result<bool, AppError> = if let (Some(domain), Some(service)) = (item.action_ha_domain.as_deref(), item.action_ha_service.as_deref()) {
|
|
let settings = state.settings.read().await.clone();
|
|
match home_assistant::call_service(&state.http, &settings.home_assistant, domain, service, item.action_ha_entity_id.as_deref(), &item.action_ha_data).await {
|
|
Ok(_) => Ok(true),
|
|
Err(err) => Err(AppError::BadRequest(format!("Home Assistant service action failed: {err}"))),
|
|
}
|
|
} else if let Some(zone_id) = item.action_zone_id.as_deref() {
|
|
apply_flow_zone_action(state, zone_id, item.action_zone_preset.as_deref(), &item.action).await
|
|
} else if let Some(group_id) = item.action_group_id.as_deref() {
|
|
let group_mode = item.action.mode.as_deref().map(|mode| if mode == "auto" { "house".to_string() } else { mode.to_string() });
|
|
control_group(state, group_id, GroupControlPatch {
|
|
power: item.action.power,
|
|
mode: group_mode,
|
|
preset: item.action_preset.clone(),
|
|
setpoint: item.action.target_temperature,
|
|
}, "automation.group").await.map(|value| !value.get("suppressed").and_then(Value::as_bool).unwrap_or(false))
|
|
} else {
|
|
match apply_automatic_device_action(state, &item.action_device_id, item.action.clone()).await {
|
|
Ok(Some(_)) => Ok(true),
|
|
Ok(None) => {
|
|
state.log("info", "automation.blocked_by_fresh_ownership", &format!("Automation {} was suppressed after ownership changed", item.name), json!({
|
|
"automation_id": item.id, "device_id": item.action_device_id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id
|
|
}));
|
|
Ok(false)
|
|
}
|
|
Err(err) => Err(err),
|
|
}
|
|
};
|
|
match result {
|
|
Ok(true) => {
|
|
for device_id in target_devices { claimed_devices.insert(device_id); }
|
|
let fired_at = Utc::now();
|
|
flow_record_rate_limited_execution(&mut item, fired_at.clone());
|
|
item.last_fired_at = Some(fired_at);
|
|
item.updated_at = Utc::now();
|
|
state.db.save_automation(&item)?;
|
|
// Keep connected clients in sync with runtime metadata such as last_fired_at.
|
|
// automation.updated also invalidates the materialized control plan.
|
|
state.broadcast("automation.updated", serde_json::to_value(&item)?);
|
|
state.log("info", "automation.fired", &format!("Automation {} fired", item.name), json!({
|
|
"automation_id": item.id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id, "group_id": item.action_group_id, "device_id": item.action_device_id
|
|
}));
|
|
}
|
|
Ok(false) => {
|
|
// Ownership suppression is not an execution. Do not consume cooldown (M3),
|
|
// so a still-valid trigger may run as soon as the higher-priority owner leaves.
|
|
if item.flow_id.is_some() {
|
|
state.log("info", "flow.action_suppressed", &format!("Flow action {} was suppressed by current ownership", item.name), json!({
|
|
"automation_id": item.id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id,
|
|
"group_id": item.action_group_id, "zone_id": item.action_zone_id, "device_id": item.action_device_id
|
|
}));
|
|
}
|
|
}
|
|
Err(err) => {
|
|
// A failed action is still an execution attempt. Apply the configured cooldown
|
|
// so an offline/disabled target cannot be hammered on every automation cycle.
|
|
item.last_fired_at = Some(Utc::now());
|
|
item.updated_at = Utc::now();
|
|
state.db.save_automation(&item)?;
|
|
state.broadcast("automation.updated", serde_json::to_value(&item)?);
|
|
state.log("error", "automation.error", &err.to_string(), json!({"automation_id": item.id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id}));
|
|
}
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn device_blocked_by_disabled_zone(device_id: &str, zones: &[Zone]) -> bool {
|
|
zones.iter().any(|zone| zone.device_id == device_id && !zone.enabled)
|
|
}
|
|
|
|
fn device_blocked_by_manual_override(device_id: &str, zones: &[Zone]) -> bool {
|
|
zones.iter().any(|zone| zone.device_id == device_id && zone.device_manual_override)
|
|
}
|
|
|
|
fn device_blocked_by_local_thermostat(device_id: &str, zones: &[Zone]) -> bool {
|
|
zones.iter().any(|zone| zone.device_id == device_id && zone.local_thermostat_power.is_some())
|
|
}
|
|
|
|
fn device_blocked_by_temporary_thermostat(device_id: &str, zones: &[Zone]) -> bool {
|
|
let now = Utc::now();
|
|
zones.iter().any(|zone| zone.device_id == device_id && temporary_quick_thermostat_is_active(zone, now.clone()))
|
|
}
|
|
|
|
fn find_temperature(devices: &[Device], device_id: Option<&str>) -> Option<f64> {
|
|
let id = device_id?;
|
|
// Never fire a temperature automation from stale cached data of an offline/disabled unit.
|
|
devices.iter().find(|d| d.id == id && d.enabled && d.online && d.communication_failures == 0)?.current_temperature
|
|
}
|
|
|
|
fn automation_ready(item: &Automation) -> bool {
|
|
item.last_fired_at.map(|last| (Utc::now() - last).num_seconds().max(0) as u64 >= item.cooldown_seconds).unwrap_or(true)
|
|
}
|
|
|
|
fn time_automation_due(item: &Automation, now: DateTime<Local>) -> bool {
|
|
let Some(expected) = item.at_time.as_deref() else { return false; };
|
|
let Ok(value) = NaiveTime::parse_from_str(expected, "%H:%M") else { return false; };
|
|
if now.hour() != value.hour() || now.minute() != value.minute() { return false; }
|
|
if let Some(last) = item.last_fired_at {
|
|
let local_last = last.with_timezone(&Local);
|
|
if local_last.date_naive() == now.date_naive()
|
|
&& local_last.hour() == now.hour()
|
|
&& local_last.minute() == now.minute()
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
true
|
|
}
|
|
|
|
|
|
|
|
fn flow_compare(actual: f64, operator: &str, expected: f64) -> bool {
|
|
match operator { "lt" => actual < expected, "lte" => actual <= expected, "gt" => actual > expected, "gte" => actual >= expected, "eq" => (actual - expected).abs() < 0.0001, "neq" => (actual - expected).abs() >= 0.0001, _ => false }
|
|
}
|
|
|
|
fn flow_value_text(value: &Value) -> String {
|
|
match value {
|
|
Value::String(v) => v.clone(),
|
|
Value::Bool(v) => v.to_string(),
|
|
Value::Number(v) => v.to_string(),
|
|
Value::Null => "null".into(),
|
|
other => other.to_string(),
|
|
}
|
|
}
|
|
|
|
fn flow_compare_value(actual: &Value, operator: &str, expected: &Value) -> bool {
|
|
if let (Some(a), Some(e)) = (actual.as_f64(), expected.as_f64()) {
|
|
return flow_compare(a, operator, e);
|
|
}
|
|
if matches!(operator, "lt" | "lte" | "gt" | "gte") {
|
|
let a = flow_value_text(actual).parse::<f64>().ok();
|
|
let e = flow_value_text(expected).parse::<f64>().ok();
|
|
return a.zip(e).map(|(a, e)| flow_compare(a, operator, e)).unwrap_or(false);
|
|
}
|
|
let equal = flow_value_text(actual).eq_ignore_ascii_case(&flow_value_text(expected));
|
|
if operator == "neq" { !equal } else { equal }
|
|
}
|
|
|
|
fn flow_device_state_value(device: &Device, field: &str) -> Option<Value> {
|
|
Some(match field {
|
|
"enabled" => json!(device.enabled),
|
|
"online" => json!(device.online),
|
|
"power" => json!(device.power),
|
|
"mode" => json!(device.mode),
|
|
"fan_speed" => json!(device.fan_speed),
|
|
"swing_vertical" => json!(device.swing_vertical),
|
|
"swing_horizontal" => json!(device.swing_horizontal),
|
|
"quiet" => json!(device.quiet),
|
|
"turbo" => json!(device.turbo),
|
|
"light" => json!(device.light),
|
|
"air" => json!(device.air),
|
|
"xfan" => json!(device.xfan),
|
|
"health" => json!(device.health),
|
|
"sleep" => json!(device.sleep),
|
|
_ => return None,
|
|
})
|
|
}
|
|
|
|
fn flow_zone_state_value(zone: &Zone, field: &str) -> Option<Value> {
|
|
Some(match field {
|
|
"enabled" => json!(zone.enabled),
|
|
"mode" => json!(zone.mode),
|
|
"active_preset" => json!(zone.active_preset),
|
|
"demand" => json!(zone.demand),
|
|
"control_owner" => json!(zone.control_owner),
|
|
"device_manual_override" => json!(zone.device_manual_override),
|
|
"local_thermostat_power" => zone.local_thermostat_power.map(Value::Bool).unwrap_or(Value::Null),
|
|
_ => return None,
|
|
})
|
|
}
|
|
|
|
|
|
fn cron_field_matches(field: &str, value: u32, min: u32, max: u32) -> bool {
|
|
fn part_matches(part: &str, value: u32, min: u32, max: u32) -> bool {
|
|
if part == "*" { return true; }
|
|
if let Some(step) = part.strip_prefix("*/").and_then(|v| v.parse::<u32>().ok()).filter(|v| *v > 0) {
|
|
return value >= min && value <= max && (value - min) % step == 0;
|
|
}
|
|
if let Some((a,b)) = part.split_once('-') {
|
|
if let (Ok(a), Ok(b)) = (a.parse::<u32>(), b.parse::<u32>()) { return a <= value && value <= b && a >= min && b <= max; }
|
|
return false;
|
|
}
|
|
part.parse::<u32>().ok().is_some_and(|v| v == value && v >= min && v <= max)
|
|
}
|
|
field.split(',').any(|part| part_matches(part.trim(), value, min, max))
|
|
}
|
|
|
|
pub(crate) fn cron_expression_valid(expression: &str) -> bool {
|
|
fn valid_field(field: &str, min: u32, max: u32, allow_seven: bool) -> bool {
|
|
let upper = if allow_seven { 7 } else { max };
|
|
if field.trim().is_empty() { return false; }
|
|
field.split(',').all(|part| {
|
|
let part = part.trim();
|
|
if part == "*" { return true; }
|
|
if let Some(raw) = part.strip_prefix("*/") {
|
|
return raw.parse::<u32>().ok().is_some_and(|step| step > 0 && step <= (max - min + 1));
|
|
}
|
|
if let Some((a, b)) = part.split_once('-') {
|
|
return a.parse::<u32>().ok().zip(b.parse::<u32>().ok())
|
|
.is_some_and(|(a, b)| a >= min && b <= upper && a <= b);
|
|
}
|
|
part.parse::<u32>().ok().is_some_and(|value| value >= min && value <= upper)
|
|
})
|
|
}
|
|
let fields: Vec<_> = expression.split_whitespace().collect();
|
|
fields.len() == 5
|
|
&& valid_field(fields[0], 0, 59, false)
|
|
&& valid_field(fields[1], 0, 23, false)
|
|
&& valid_field(fields[2], 1, 31, false)
|
|
&& valid_field(fields[3], 1, 12, false)
|
|
&& valid_field(fields[4], 0, 6, true)
|
|
}
|
|
|
|
fn cron_matches(expression: &str, now: &DateTime<Local>) -> bool {
|
|
if !cron_expression_valid(expression) { return false; }
|
|
let fields: Vec<_> = expression.split_whitespace().collect();
|
|
let weekday = now.weekday().num_days_from_sunday();
|
|
cron_field_matches(fields[0], now.minute(), 0, 59)
|
|
&& cron_field_matches(fields[1], now.hour(), 0, 23)
|
|
&& cron_field_matches(fields[2], now.day(), 1, 31)
|
|
&& cron_field_matches(fields[3], now.month(), 1, 12)
|
|
&& (cron_field_matches(fields[4], weekday, 0, 7) || (weekday == 0 && cron_field_matches(fields[4], 7, 0, 7)))
|
|
}
|
|
|
|
fn flow_oscillation_metrics(samples: &[crate::models::FlowRuntimeSample]) -> Option<(f64, usize)> {
|
|
if samples.len() < 3 { return None; }
|
|
let min = samples.iter().map(|item| item.value).fold(f64::INFINITY, f64::min);
|
|
let max = samples.iter().map(|item| item.value).fold(f64::NEG_INFINITY, f64::max);
|
|
if !min.is_finite() || !max.is_finite() { return None; }
|
|
let mut previous_sign = 0i8;
|
|
let mut direction_changes = 0usize;
|
|
for pair in samples.windows(2) {
|
|
let delta = pair[1].value - pair[0].value;
|
|
let sign = if delta > 0.000001 { 1 } else if delta < -0.000001 { -1 } else { 0 };
|
|
if sign == 0 { continue; }
|
|
if previous_sign != 0 && sign != previous_sign { direction_changes += 1; }
|
|
previous_sign = sign;
|
|
}
|
|
Some((max - min, direction_changes))
|
|
}
|
|
|
|
fn flow_timed_gate_update(
|
|
state: &mut crate::models::FlowRuntimeNodeState,
|
|
input: bool,
|
|
seconds: u64,
|
|
now: DateTime<Utc>,
|
|
) -> bool {
|
|
if !input { state.since = None; return false; }
|
|
let since = state.since.get_or_insert_with(|| now.clone());
|
|
now.signed_duration_since(since.clone()).num_seconds() >= seconds as i64
|
|
}
|
|
|
|
fn flow_state_duration_update(
|
|
state: &mut crate::models::FlowRuntimeNodeState,
|
|
input: bool,
|
|
min_seconds: u64,
|
|
max_seconds: Option<u64>,
|
|
now: DateTime<Utc>,
|
|
) -> (bool, u64) {
|
|
if !input { state.since = None; return (false, 0); }
|
|
let since = state.since.get_or_insert_with(|| now.clone());
|
|
let elapsed = now.signed_duration_since(since.clone()).num_seconds().max(0) as u64;
|
|
let within_min = elapsed >= min_seconds;
|
|
let within_max = max_seconds.map(|max| elapsed <= max).unwrap_or(true);
|
|
(within_min && within_max, elapsed)
|
|
}
|
|
|
|
fn flow_change_gate_update(state: &mut crate::models::FlowRuntimeNodeState, current: Value) -> bool {
|
|
let changed = state.last_value.as_ref().is_some_and(|previous| previous != ¤t);
|
|
state.last_value = Some(current);
|
|
changed
|
|
}
|
|
|
|
fn flow_rate_limit_status(
|
|
state: &mut crate::models::FlowRuntimeNodeState,
|
|
max_count: usize,
|
|
period_seconds: u64,
|
|
now: DateTime<Utc>,
|
|
) -> (bool, usize) {
|
|
let cutoff = now - chrono::Duration::seconds(period_seconds as i64);
|
|
state.samples.retain(|item| item.at >= cutoff);
|
|
let used = state.samples.len();
|
|
(used < max_count, used)
|
|
}
|
|
|
|
fn flow_record_rate_limited_execution(item: &mut Automation, now: DateTime<Utc>) {
|
|
let rate_limits = item.flow_conditions.iter()
|
|
.filter(|condition| condition.kind == "rate_limit")
|
|
.filter(|condition| item.flow_runtime.get(&condition.id).and_then(|state| state.last_value.as_ref()).and_then(Value::as_bool) == Some(true))
|
|
.map(|condition| (condition.id.clone(), condition.config.get("period_seconds").and_then(Value::as_u64).unwrap_or(3600)))
|
|
.collect::<Vec<_>>();
|
|
for (node_id, period_seconds) in rate_limits {
|
|
let state = item.flow_runtime.entry(node_id).or_default();
|
|
let cutoff = now.clone() - chrono::Duration::seconds(period_seconds as i64);
|
|
state.samples.retain(|sample| sample.at >= cutoff);
|
|
state.samples.push(crate::models::FlowRuntimeSample { at: now.clone(), value: 1.0 });
|
|
}
|
|
}
|
|
|
|
fn flow_rolling_stat_update(
|
|
state: &mut crate::models::FlowRuntimeNodeState,
|
|
sample: f64,
|
|
window_seconds: u64,
|
|
statistic: &str,
|
|
now: DateTime<Utc>,
|
|
) -> Option<f64> {
|
|
let cutoff = now.clone() - chrono::Duration::seconds(window_seconds as i64);
|
|
state.samples.retain(|item| item.at >= cutoff);
|
|
state.samples.push(crate::models::FlowRuntimeSample { at: now, value: sample });
|
|
let mut values = state.samples.iter().map(|item| item.value).collect::<Vec<_>>();
|
|
if values.is_empty() { return None; }
|
|
if statistic == "median" {
|
|
values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
|
|
let mid = values.len() / 2;
|
|
Some(if values.len() % 2 == 0 { (values[mid - 1] + values[mid]) / 2.0 } else { values[mid] })
|
|
} else {
|
|
Some(values.iter().sum::<f64>() / values.len() as f64)
|
|
}
|
|
}
|
|
|
|
async fn flow_leaf_observation(
|
|
state: &AppState,
|
|
devices: &[Device],
|
|
zones: &[Zone],
|
|
outdoor_temperature: Option<f64>,
|
|
condition: &crate::models::FlowCondition,
|
|
now: &DateTime<Local>,
|
|
settings: &crate::models::RuntimeSettings,
|
|
overrides: &HashMap<String, Value>,
|
|
) -> Result<(bool, Value), AppError> {
|
|
let mut resolved_kind = condition.kind.as_str();
|
|
let mut resolved_config: Option<Value> = None;
|
|
if condition.kind == "shared_input" {
|
|
let input_id = condition.config.get("input_id").and_then(Value::as_str).unwrap_or("");
|
|
let Some(shared) = settings.home_assistant.flow_inputs.iter().find(|item| item.id == input_id) else {
|
|
return Ok((false, json!({"error": "missing_shared_input", "input_id": input_id})));
|
|
};
|
|
resolved_kind = shared.kind.as_str();
|
|
let mut config = shared.config.clone();
|
|
if matches!(shared.kind.as_str(),
|
|
"outdoor_temperature" | "device_temperature" | "zone_temperature" |
|
|
"ha_state" | "ha_numeric" | "ha_attribute" | "house_mode" |
|
|
"device_state" | "zone_state" | "group_state") {
|
|
let Some(operator) = condition.config.get("operator").and_then(Value::as_str).map(str::trim).filter(|value| !value.is_empty()) else {
|
|
return Ok((false, json!({"error": "missing_shared_input_operator", "input_id": input_id})));
|
|
};
|
|
let Some(map) = config.as_object_mut() else {
|
|
return Ok((false, json!({"error": "invalid_shared_input_config", "input_id": input_id})));
|
|
};
|
|
map.insert("operator".into(), Value::String(operator.to_string()));
|
|
map.insert("value".into(), condition.config.get("value").cloned().unwrap_or(Value::Null));
|
|
}
|
|
resolved_config = Some(config);
|
|
}
|
|
let c = resolved_config.as_ref().unwrap_or(&condition.config);
|
|
let override_value = overrides.get(&condition.id).cloned();
|
|
let (matched, actual) = match resolved_kind {
|
|
"weekday" => {
|
|
let actual = override_value.unwrap_or_else(|| json!(now.weekday().number_from_monday()));
|
|
let day = actual.as_u64().unwrap_or(now.weekday().number_from_monday() as u64);
|
|
let matched = c.get("days").and_then(Value::as_array)
|
|
.map(|days| days.iter().filter_map(Value::as_u64).any(|expected| expected == day)).unwrap_or(false);
|
|
(matched, json!(day))
|
|
}
|
|
"time_range" => {
|
|
let actual = override_value.and_then(|v| v.as_str().map(str::to_string)).unwrap_or_else(|| now.format("%H:%M").to_string());
|
|
let current = NaiveTime::parse_from_str(&actual, "%H:%M").unwrap_or_else(|_| now.time());
|
|
let start = c.get("start").and_then(Value::as_str).and_then(|v| NaiveTime::parse_from_str(v, "%H:%M").ok());
|
|
let end = c.get("end").and_then(Value::as_str).and_then(|v| NaiveTime::parse_from_str(v, "%H:%M").ok());
|
|
let matched = match (start, end) {
|
|
(Some(start), Some(end)) if start == end => true,
|
|
(Some(start), Some(end)) if start < end => current >= start && current < end,
|
|
(Some(start), Some(end)) => current >= start || current < end,
|
|
_ => false,
|
|
};
|
|
(matched, json!(actual))
|
|
}
|
|
"date_range" => {
|
|
let actual = override_value.and_then(|v| v.as_str().map(str::to_string)).unwrap_or_else(|| now.date_naive().format("%Y-%m-%d").to_string());
|
|
let current = chrono::NaiveDate::parse_from_str(&actual, "%Y-%m-%d").unwrap_or_else(|_| now.date_naive());
|
|
let start = c.get("start").and_then(Value::as_str).and_then(|v| chrono::NaiveDate::parse_from_str(v, "%Y-%m-%d").ok());
|
|
let end = c.get("end").and_then(Value::as_str).and_then(|v| chrono::NaiveDate::parse_from_str(v, "%Y-%m-%d").ok());
|
|
(matches!((start, end), (Some(a), Some(b)) if current >= a && current <= b), json!(actual))
|
|
}
|
|
"cron_trigger" => {
|
|
let expression = c.get("expression").and_then(Value::as_str).unwrap_or("");
|
|
(cron_matches(expression, now), json!(now.format("%Y-%m-%d %H:%M").to_string()))
|
|
}
|
|
"outdoor_temperature" => {
|
|
let actual = if let Some(v) = override_value { v.as_f64() } else { outdoor_temperature };
|
|
let expected = c.get("value").and_then(Value::as_f64);
|
|
(actual.zip(expected).map(|(a, e)| flow_compare(a, c.get("operator").and_then(Value::as_str).unwrap_or("lt"), e)).unwrap_or(false), actual.map_or(Value::Null, |v| json!(v)))
|
|
}
|
|
"device_temperature" => {
|
|
let id = c.get("device_id").and_then(Value::as_str).unwrap_or("");
|
|
let actual = if let Some(v) = override_value { v.as_f64() } else { find_temperature(devices, Some(id)) };
|
|
let expected = c.get("value").and_then(Value::as_f64);
|
|
(actual.zip(expected).map(|(a, e)| flow_compare(a, c.get("operator").and_then(Value::as_str).unwrap_or("lt"), e)).unwrap_or(false), actual.map_or(Value::Null, |v| json!(v)))
|
|
}
|
|
"zone_temperature" => {
|
|
let id = c.get("zone_id").and_then(Value::as_str).unwrap_or("");
|
|
let actual = if let Some(v) = override_value { v.as_f64() } else { zones.iter().find(|zone| zone.id == id).and_then(|zone| zone.current_temperature) };
|
|
let expected = c.get("value").and_then(Value::as_f64);
|
|
(actual.zip(expected).map(|(a, e)| flow_compare(a, c.get("operator").and_then(Value::as_str).unwrap_or("lt"), e)).unwrap_or(false), actual.map_or(Value::Null, |v| json!(v)))
|
|
}
|
|
"ha_state" => {
|
|
let entity = c.get("entity_id").and_then(Value::as_str).unwrap_or("");
|
|
let actual = match override_value {
|
|
Some(value) => value,
|
|
None => match home_assistant::read_state(&state.http, &settings.home_assistant, Some(entity)).await {
|
|
Ok(value) => json!(value),
|
|
Err(err) => return Ok((false, json!({"error": err.to_string(), "entity_id": entity}))),
|
|
},
|
|
};
|
|
let expected = c.get("value").cloned().unwrap_or(Value::Null);
|
|
(flow_compare_value(&actual, c.get("operator").and_then(Value::as_str).unwrap_or("eq"), &expected), actual)
|
|
}
|
|
"ha_numeric" => {
|
|
let entity = c.get("entity_id").and_then(Value::as_str).unwrap_or("");
|
|
let actual = match override_value {
|
|
Some(value) => value,
|
|
None => {
|
|
let raw = match home_assistant::read_state(&state.http, &settings.home_assistant, Some(entity)).await {
|
|
Ok(value) => value,
|
|
Err(err) => return Ok((false, json!({"error": err.to_string(), "entity_id": entity}))),
|
|
};
|
|
match raw.parse::<f64>() {
|
|
Ok(value) => json!(value),
|
|
Err(_) => return Ok((false, json!({"error": "non_numeric_state", "entity_id": entity, "state": raw}))),
|
|
}
|
|
}
|
|
};
|
|
let expected = c.get("value").cloned().unwrap_or(Value::Null);
|
|
(flow_compare_value(&actual, c.get("operator").and_then(Value::as_str).unwrap_or("lt"), &expected), actual)
|
|
}
|
|
"ha_attribute" => {
|
|
let entity = c.get("entity_id").and_then(Value::as_str).unwrap_or("");
|
|
let attribute = c.get("attribute").and_then(Value::as_str).unwrap_or("");
|
|
let actual = match override_value {
|
|
Some(value) => value,
|
|
None => {
|
|
let payload = match home_assistant::read_entity(&state.http, &settings.home_assistant, Some(entity)).await {
|
|
Ok(value) => value,
|
|
Err(err) => return Ok((false, json!({"error": err.to_string(), "entity_id": entity, "attribute": attribute}))),
|
|
};
|
|
let Some(value) = payload.get("attributes").and_then(|attrs| attrs.get(attribute)).cloned() else {
|
|
return Ok((false, json!({"error": "missing_attribute", "entity_id": entity, "attribute": attribute})));
|
|
};
|
|
value
|
|
}
|
|
};
|
|
let expected = c.get("value").cloned().unwrap_or(Value::Null);
|
|
(flow_compare_value(&actual, c.get("operator").and_then(Value::as_str).unwrap_or("eq"), &expected), actual)
|
|
}
|
|
"ha_available" => {
|
|
let entity = c.get("entity_id").and_then(Value::as_str).unwrap_or("");
|
|
let actual = match override_value {
|
|
Some(value) => value,
|
|
None => match home_assistant::read_state(&state.http, &settings.home_assistant, Some(entity)).await {
|
|
Ok(value) => Value::String(value),
|
|
Err(err) => return Ok((false, json!({"error": err.to_string(), "entity_id": entity}))),
|
|
},
|
|
};
|
|
let matched = actual.as_bool().unwrap_or_else(|| actual.as_str().map(|value| {
|
|
let value = value.trim();
|
|
!value.is_empty() && !value.eq_ignore_ascii_case("unknown") && !value.eq_ignore_ascii_case("unavailable")
|
|
}).unwrap_or(false));
|
|
(matched, actual)
|
|
}
|
|
"house_mode" => {
|
|
let actual = override_value.unwrap_or_else(|| json!(settings.house_mode));
|
|
let expected = c.get("value").cloned().unwrap_or(Value::Null);
|
|
(flow_compare_value(&actual, c.get("operator").and_then(Value::as_str).unwrap_or("eq"), &expected), actual)
|
|
}
|
|
"device_state" => {
|
|
let id = c.get("device_id").and_then(Value::as_str).unwrap_or("");
|
|
let field = c.get("field").and_then(Value::as_str).unwrap_or("");
|
|
let actual = override_value.unwrap_or_else(|| devices.iter().find(|device| device.id == id).and_then(|device| flow_device_state_value(device, field)).unwrap_or(Value::Null));
|
|
let expected = c.get("value").cloned().unwrap_or(Value::Null);
|
|
(flow_compare_value(&actual, c.get("operator").and_then(Value::as_str).unwrap_or("eq"), &expected), actual)
|
|
}
|
|
"zone_state" => {
|
|
let id = c.get("zone_id").and_then(Value::as_str).unwrap_or("");
|
|
let field = c.get("field").and_then(Value::as_str).unwrap_or("");
|
|
let actual = override_value.unwrap_or_else(|| zones.iter().find(|zone| zone.id == id).and_then(|zone| flow_zone_state_value(zone, field)).unwrap_or(Value::Null));
|
|
let expected = c.get("value").cloned().unwrap_or(Value::Null);
|
|
(flow_compare_value(&actual, c.get("operator").and_then(Value::as_str).unwrap_or("eq"), &expected), actual)
|
|
}
|
|
"group_state" => {
|
|
let id = c.get("group_id").and_then(Value::as_str).unwrap_or("");
|
|
let field = c.get("field").and_then(Value::as_str).unwrap_or("");
|
|
let actual = override_value.unwrap_or_else(|| state.db.get_group(id).ok().flatten().and_then(|group| match field {
|
|
"power_enabled" => Some(json!(group.power_enabled)),
|
|
_ => None,
|
|
}).unwrap_or(Value::Null));
|
|
let expected = c.get("value").cloned().unwrap_or(Value::Null);
|
|
(flow_compare_value(&actual, c.get("operator").and_then(Value::as_str).unwrap_or("eq"), &expected), actual)
|
|
}
|
|
"night_mode" => {
|
|
let actual = override_value.unwrap_or_else(|| json!(night_mode_active(&settings.night_mode, now.time())));
|
|
(actual.as_bool().unwrap_or(false), actual)
|
|
}
|
|
"constant" => {
|
|
let actual = override_value.unwrap_or_else(|| c.get("value").cloned().unwrap_or(Value::Bool(false)));
|
|
(actual.as_bool().unwrap_or(false), actual)
|
|
}
|
|
_ => (false, Value::Null),
|
|
};
|
|
Ok((matched, actual))
|
|
}
|
|
|
|
pub async fn evaluate_flow_conditions_trace(
|
|
state: &AppState,
|
|
devices: &[Device],
|
|
conditions: &[crate::models::FlowCondition],
|
|
now: DateTime<Local>,
|
|
overrides: &HashMap<String, Value>,
|
|
mut runtime: Option<&mut std::collections::BTreeMap<String, crate::models::FlowRuntimeNodeState>>,
|
|
) -> Result<(bool, Vec<Value>), AppError> {
|
|
if conditions.is_empty() { return Ok((false, Vec::new())); }
|
|
let settings = state.settings.read().await.clone();
|
|
let zones = state.db.list_zones()?;
|
|
let outdoor_temperature = *state.outdoor_temperature.read().await;
|
|
let mut trace = Vec::new();
|
|
|
|
if conditions.iter().all(|condition| condition.id.is_empty()) {
|
|
let mut final_value = true;
|
|
for condition in conditions {
|
|
let (matched, actual) = flow_leaf_observation(state, devices, &zones, outdoor_temperature, condition, &now, &settings, overrides).await?;
|
|
final_value &= matched;
|
|
trace.push(json!({"node_id": condition.id, "kind": condition.kind, "matched": matched, "actual": actual, "expected": condition.config.get("value")}));
|
|
}
|
|
return Ok((final_value, trace));
|
|
}
|
|
|
|
let mut values = HashMap::<String, bool>::new();
|
|
let mut actual_values = HashMap::<String, Value>::new();
|
|
let mut final_id = None::<String>;
|
|
for condition in conditions {
|
|
if condition.id.is_empty() { return Ok((false, trace)); }
|
|
let (matched, actual) = match condition.kind.as_str() {
|
|
"logic_and" => {
|
|
let value = !condition.inputs.is_empty() && condition.inputs.iter().all(|id| values.get(id).copied().unwrap_or(false));
|
|
(value, json!(condition.inputs.iter().map(|id| values.get(id).copied().unwrap_or(false)).collect::<Vec<_>>()))
|
|
}
|
|
"logic_or" => {
|
|
let value = !condition.inputs.is_empty() && condition.inputs.iter().any(|id| values.get(id).copied().unwrap_or(false));
|
|
(value, json!(condition.inputs.iter().map(|id| values.get(id).copied().unwrap_or(false)).collect::<Vec<_>>()))
|
|
}
|
|
"logic_not" => {
|
|
let value = condition.inputs.len() == 1 && !values.get(&condition.inputs[0]).copied().unwrap_or(false);
|
|
(value, json!(condition.inputs.first().and_then(|id| values.get(id)).copied()))
|
|
}
|
|
"stable_for" | "delay" => {
|
|
let input = condition.inputs.len() == 1 && values.get(&condition.inputs[0]).copied().unwrap_or(false);
|
|
let seconds = condition.config.get("seconds").and_then(Value::as_u64).unwrap_or(1);
|
|
let mut since_value = None;
|
|
let value = if let Some(map) = runtime.as_deref_mut() {
|
|
let state = map.entry(condition.id.clone()).or_default();
|
|
let value = flow_timed_gate_update(state, input, seconds, now.with_timezone(&Utc));
|
|
since_value = state.since.clone();
|
|
value
|
|
} else { false };
|
|
(value, json!({"input": input, "since": since_value, "seconds": seconds}))
|
|
}
|
|
"state_duration" => {
|
|
let input = condition.inputs.len() == 1 && values.get(&condition.inputs[0]).copied().unwrap_or(false);
|
|
let min_seconds = condition.config.get("min_seconds").and_then(Value::as_u64).unwrap_or(0);
|
|
let max_seconds = condition.config.get("max_seconds").and_then(Value::as_u64);
|
|
let mut since_value = None;
|
|
let mut elapsed_seconds = 0u64;
|
|
let value = if let Some(map) = runtime.as_deref_mut() {
|
|
let state = map.entry(condition.id.clone()).or_default();
|
|
let (value, elapsed) = flow_state_duration_update(state, input, min_seconds, max_seconds, now.with_timezone(&Utc));
|
|
since_value = state.since.clone();
|
|
elapsed_seconds = elapsed;
|
|
value
|
|
} else { false };
|
|
(value, json!({"input": input, "since": since_value, "elapsed_seconds": elapsed_seconds, "min_seconds": min_seconds, "max_seconds": max_seconds}))
|
|
}
|
|
"on_change" => {
|
|
let input_id = condition.inputs.first().cloned().unwrap_or_default();
|
|
let input = condition.inputs.len() == 1 && values.get(&input_id).copied().unwrap_or(false);
|
|
let mode = condition.config.get("mode").and_then(Value::as_str).unwrap_or("result");
|
|
let observed = if mode == "value" { actual_values.get(&input_id).cloned() } else { Some(json!(input)) };
|
|
let mut changed = false;
|
|
let mut previous = None;
|
|
if let (Some(current), Some(map)) = (observed.clone(), runtime.as_deref_mut()) {
|
|
let state = map.entry(condition.id.clone()).or_default();
|
|
previous = state.last_value.clone();
|
|
changed = flow_change_gate_update(state, current);
|
|
}
|
|
(input && changed, json!({"input": input, "mode": mode, "previous": previous, "current": observed, "changed": changed}))
|
|
}
|
|
"rate_limit" => {
|
|
let input = condition.inputs.len() == 1 && values.get(&condition.inputs[0]).copied().unwrap_or(false);
|
|
let max_count = condition.config.get("max_count").and_then(Value::as_u64).unwrap_or(1) as usize;
|
|
let period_seconds = condition.config.get("period_seconds").and_then(Value::as_u64).unwrap_or(3600);
|
|
let mut used = 0usize;
|
|
let available = if let Some(map) = runtime.as_deref_mut() {
|
|
let state = map.entry(condition.id.clone()).or_default();
|
|
let (available, count) = flow_rate_limit_status(state, max_count, period_seconds, now.with_timezone(&Utc));
|
|
used = count;
|
|
state.last_value = Some(json!(input && available));
|
|
available
|
|
} else { false };
|
|
(input && available, json!({"input": input, "used": used, "max_count": max_count, "period_seconds": period_seconds, "remaining": max_count.saturating_sub(used)}))
|
|
}
|
|
"rolling_stat" => {
|
|
let predecessors_match = condition.inputs.iter().all(|id| values.get(id).copied().unwrap_or(false));
|
|
let source = condition.config.get("source").and_then(Value::as_str).unwrap_or("");
|
|
let sample = if let Some(value) = overrides.get(&condition.id).and_then(Value::as_f64) {
|
|
Some(value)
|
|
} else {
|
|
match source {
|
|
"outdoor_temperature" => outdoor_temperature,
|
|
"device_temperature" => condition.config.get("device_id").and_then(Value::as_str).and_then(|id| find_temperature(devices, Some(id))),
|
|
"zone_temperature" => condition.config.get("zone_id").and_then(Value::as_str).and_then(|id| zones.iter().find(|zone| zone.id == id).and_then(|zone| zone.current_temperature)),
|
|
"ha_numeric" => {
|
|
let entity = condition.config.get("entity_id").and_then(Value::as_str).unwrap_or("");
|
|
match home_assistant::read_state(&state.http, &settings.home_assistant, Some(entity)).await { Ok(raw) => raw.parse::<f64>().ok(), Err(_) => None }
|
|
}
|
|
_ => None,
|
|
}
|
|
};
|
|
let window = condition.config.get("window_seconds").and_then(Value::as_u64).unwrap_or(60);
|
|
let mut aggregate = None;
|
|
let mut count = 0usize;
|
|
if let (Some(sample), Some(map)) = (sample, runtime.as_deref_mut()) {
|
|
let node_state = map.entry(condition.id.clone()).or_default();
|
|
aggregate = flow_rolling_stat_update(
|
|
node_state, sample, window, condition.config.get("statistic").and_then(Value::as_str).unwrap_or("mean"), now.with_timezone(&Utc),
|
|
);
|
|
count = node_state.samples.len();
|
|
}
|
|
let expected = condition.config.get("value").and_then(Value::as_f64);
|
|
let matched = predecessors_match && aggregate.zip(expected).map(|(a,e)| flow_compare(a, condition.config.get("operator").and_then(Value::as_str).unwrap_or("lt"), e)).unwrap_or(false);
|
|
(matched, json!({"sample": sample, "aggregate": aggregate, "samples": count, "window_seconds": window, "statistic": condition.config.get("statistic")}))
|
|
}
|
|
"oscillates" => {
|
|
let predecessors_match = condition.inputs.iter().all(|id| values.get(id).copied().unwrap_or(false));
|
|
let source = condition.config.get("source").and_then(Value::as_str).unwrap_or("");
|
|
let sample = if let Some(value) = overrides.get(&condition.id).and_then(Value::as_f64) {
|
|
Some(value)
|
|
} else {
|
|
match source {
|
|
"outdoor_temperature" => outdoor_temperature,
|
|
"device_temperature" => condition.config.get("device_id").and_then(Value::as_str).and_then(|id| find_temperature(devices, Some(id))),
|
|
"zone_temperature" => condition.config.get("zone_id").and_then(Value::as_str).and_then(|id| zones.iter().find(|zone| zone.id == id).and_then(|zone| zone.current_temperature)),
|
|
"ha_numeric" => {
|
|
let entity = condition.config.get("entity_id").and_then(Value::as_str).unwrap_or("");
|
|
match home_assistant::read_state(&state.http, &settings.home_assistant, Some(entity)).await { Ok(raw) => raw.parse::<f64>().ok(), Err(_) => None }
|
|
}
|
|
_ => None,
|
|
}
|
|
};
|
|
let window = condition.config.get("window_seconds").and_then(Value::as_u64).unwrap_or(300);
|
|
let min_span = condition.config.get("min_span").and_then(Value::as_f64).unwrap_or(1.0);
|
|
let min_changes = condition.config.get("min_direction_changes").and_then(Value::as_u64).unwrap_or(2) as usize;
|
|
let cutoff = now.with_timezone(&Utc) - chrono::Duration::seconds(window as i64);
|
|
let mut count = 0usize;
|
|
let mut span = None;
|
|
let mut direction_changes = 0usize;
|
|
if let (Some(sample), Some(map)) = (sample, runtime.as_deref_mut()) {
|
|
let node_state = map.entry(condition.id.clone()).or_default();
|
|
node_state.samples.retain(|item| item.at >= cutoff);
|
|
node_state.samples.push(crate::models::FlowRuntimeSample { at: now.with_timezone(&Utc), value: sample });
|
|
count = node_state.samples.len();
|
|
if let Some((value_span, changes)) = flow_oscillation_metrics(&node_state.samples) {
|
|
span = Some(value_span);
|
|
direction_changes = changes;
|
|
}
|
|
}
|
|
let matched = predecessors_match
|
|
&& span.is_some_and(|value| value >= min_span)
|
|
&& direction_changes >= min_changes;
|
|
(matched, json!({"sample": sample, "samples": count, "window_seconds": window, "span": span, "min_span": min_span, "direction_changes": direction_changes, "min_direction_changes": min_changes}))
|
|
}
|
|
_ if flow_condition_kind_runtime(&condition.kind) => {
|
|
let predecessors_match = condition.inputs.iter().all(|id| values.get(id).copied().unwrap_or(false));
|
|
let (leaf, actual) = flow_leaf_observation(state, devices, &zones, outdoor_temperature, condition, &now, &settings, overrides).await?;
|
|
(predecessors_match && leaf, actual)
|
|
}
|
|
_ => (false, Value::Null),
|
|
};
|
|
values.insert(condition.id.clone(), matched);
|
|
actual_values.insert(condition.id.clone(), actual.clone());
|
|
final_id = Some(condition.id.clone());
|
|
let expected = condition.config.get("value").cloned();
|
|
trace.push(json!({
|
|
"node_id": condition.id,
|
|
"kind": condition.kind,
|
|
"matched": matched,
|
|
"actual": actual,
|
|
"expected": expected,
|
|
"inputs": condition.inputs
|
|
}));
|
|
}
|
|
Ok((final_id.and_then(|id| values.get(&id).copied()).unwrap_or(false), trace))
|
|
}
|
|
|
|
async fn flow_conditions_match(state: &AppState, devices: &[Device], item: &mut Automation) -> Result<bool, AppError> {
|
|
let overrides = HashMap::new();
|
|
let conditions = item.flow_conditions.clone();
|
|
let now = Local::now();
|
|
let (matched, _) = evaluate_flow_conditions_trace(state, devices, &conditions, now.clone(), &overrides, Some(&mut item.flow_runtime)).await?;
|
|
if matched && conditions.iter().any(|condition| condition.kind == "cron_trigger") {
|
|
if let Some(last) = item.last_fired_at.as_ref().map(|value| value.with_timezone(&Local)) {
|
|
if last.year() == now.year() && last.ordinal() == now.ordinal() && last.hour() == now.hour() && last.minute() == now.minute() { return Ok(false); }
|
|
}
|
|
}
|
|
Ok(matched)
|
|
}
|
|
|
|
fn flow_condition_kind_runtime(kind: &str) -> bool {
|
|
matches!(kind,
|
|
"weekday" | "time_range" | "date_range" | "cron_trigger" | "outdoor_temperature" | "device_temperature" |
|
|
"zone_temperature" | "ha_state" | "ha_numeric" | "ha_attribute" | "ha_available" | "house_mode" |
|
|
"device_state" | "zone_state" | "group_state" | "night_mode" | "constant" | "shared_input"
|
|
)
|
|
}
|
|
|
|
async fn apply_flow_zone_action(state: &AppState, zone_id: &str, preset: Option<&str>, action: &DeviceCommand) -> Result<bool, AppError> {
|
|
let _schedule_guard = state.lock_schedule_operation().await;
|
|
// Match the canonical thermostat ordering used by Web/HA zone control. This prevents
|
|
// a Flow action from racing a thermostat arbitration cycle that already holds a stale snapshot.
|
|
let _cycle_guard = state.lock_zone_control_cycle().await;
|
|
let _zone_guard = state.lock_zone_operation(zone_id).await;
|
|
let Some(mut zone) = state.db.get_zone(zone_id)? else { return Err(AppError::NotFound(format!("zone {zone_id}"))); };
|
|
if zone.device_manual_override || zone.local_thermostat_power.is_some() || temporary_quick_thermostat_is_active(&zone, Utc::now()) { return Ok(false); }
|
|
// Match the existing automatic-device ownership semantics: a disabled thermostat zone
|
|
// remains an explicit gate unless this Flow action is the actor re-enabling it.
|
|
if !zone.enabled && action.power != Some(true) { return Ok(false); }
|
|
let device_id = zone.device_id.clone();
|
|
if let Some(power) = action.power { zone.enabled = power; }
|
|
if let Some(mode) = action.mode.as_deref() {
|
|
match mode { "auto" => zone.inherit_house_mode = true, "heat" | "cool" => { zone.mode = mode.to_string(); zone.inherit_house_mode = false; }, _ => return Err(AppError::BadRequest("unsupported Flow thermostat mode".into())) }
|
|
}
|
|
match preset.unwrap_or("auto") {
|
|
"auto" => { zone.manual_preset = None; zone.manual_setpoint = None; zone.manual_override_until = None; }
|
|
"custom" => {
|
|
let target = action.target_temperature.ok_or_else(|| AppError::BadRequest("Flow custom thermostat action has no target".into()))?;
|
|
zone.manual_preset = Some("custom".into()); zone.manual_setpoint = Some(normalize_thermostat_target(target.clamp(8.0, 30.0)));
|
|
zone.manual_override_until = next_schedule_boundary_utc(&zone.id, &state.db.list_schedules()?, Local::now());
|
|
}
|
|
value @ ("comfort" | "sleep" | "away") => {
|
|
zone.manual_preset = Some(value.into()); zone.manual_setpoint = None;
|
|
zone.manual_override_until = next_schedule_boundary_utc(&zone.id, &state.db.list_schedules()?, Local::now());
|
|
}
|
|
_ => return Err(AppError::BadRequest("unsupported Flow thermostat preset".into())),
|
|
}
|
|
rearm_compressor_queue(&mut zone);
|
|
if action.power == Some(false) {
|
|
zone.demand = false;
|
|
zone.demand_since = None;
|
|
zone.effective_mode = "off".into();
|
|
zone.device_setpoint = None;
|
|
}
|
|
zone.revision = zone.revision.saturating_add(1);
|
|
zone.updated_at = Utc::now();
|
|
zone.control_owner = "automation".into();
|
|
zone.control_source = "automation.flow".into();
|
|
zone.control_reason = "Visual Flow automation".into();
|
|
let schedules = state.db.list_schedules()?;
|
|
let house_mode = state.settings.read().await.house_mode.clone();
|
|
refresh_control_ownership(&mut zone);
|
|
if zone.control_owner == "automation" {
|
|
zone.control_source = "automation.flow".into();
|
|
zone.control_reason = "Visual Flow automation".into();
|
|
}
|
|
refresh_zone_runtime_target(&mut zone, &schedules, &house_mode);
|
|
state.db.save_zone(&zone)?;
|
|
state.broadcast("zone.updated", serde_json::to_value(&zone)?);
|
|
|
|
let swing_command = DeviceCommand {
|
|
swing_vertical: action.swing_vertical,
|
|
swing_horizontal: action.swing_horizontal,
|
|
..Default::default()
|
|
};
|
|
if action.power == Some(false) {
|
|
// Disabled zones are intentionally skipped by the normal thermostat cycle. Perform the
|
|
// physical OFF under the canonical zone -> device lock order so the durable Flow intent
|
|
// cannot leave a unit running and polling/manual control cannot interleave with the frame.
|
|
// Swing can safely share this explicit frame because it is outside thermostat regulation.
|
|
let _device_guard = state.lock_device_operation(&device_id).await;
|
|
send_command_locked_forced(
|
|
state,
|
|
&device_id,
|
|
DeviceCommand {
|
|
power: Some(false),
|
|
swing_vertical: swing_command.swing_vertical,
|
|
swing_horizontal: swing_command.swing_horizontal,
|
|
..Default::default()
|
|
},
|
|
)
|
|
.await?;
|
|
} else if !swing_command.is_empty() {
|
|
// Swing is intentionally a one-shot auxiliary unit setting. It does not participate in
|
|
// temperature/fan regulation, so the thermostat keeps ownership of the zone.
|
|
let _ = send_automatic_device_command_if_owned(state, &device_id, swing_command).await?;
|
|
}
|
|
state.wake_zone_control();
|
|
Ok(true)
|
|
}
|