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

v0.11.0

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-09-02 22:41:27 +02:00
parent 8004be0841
commit db6d2f09db
20 changed files with 945 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/automations.rs
+1 -1
View File
@@ -87,7 +87,7 @@ impl AutomationInput {
action_group_id: self.action_group_id.map(|value| value.trim().to_string()).filter(|value| !value.is_empty()),
action_preset: self.action_preset.map(|value| value.trim().to_string()).filter(|value| !value.is_empty()),
action: self.action, cooldown_seconds: self.cooldown_seconds.max(30), last_fired_at,
action_zone_id: None, action_zone_preset: None, flow_conditions: vec![], flow_id: None, flow_node_id: None,
action_zone_id: None, action_zone_preset: None, action_ha_domain: None, action_ha_service: None, action_ha_entity_id: None, action_ha_data: Value::Null, flow_conditions: vec![], flow_id: None, flow_node_id: None, flow_runtime: Default::default(),
created_at, updated_at: Utc::now() }
}
}
src/api/flows.rs
+69 -5
View File
@@ -43,13 +43,13 @@ fn generated_flow_name(flow_id: &str, action_node_id: &str) -> String {
fn flow_condition_kind(kind: &str) -> bool {
matches!(kind,
"weekday" | "time_range" | "date_range" | "outdoor_temperature" | "device_temperature" |
"weekday" | "time_range" | "date_range" | "cron_trigger" | "stable_for" | "delay" | "rolling_stat" | "oscillates" | "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"
)
}
fn flow_logic_kind(kind: &str) -> bool { matches!(kind, "logic_and" | "logic_or" | "logic_not") }
fn flow_action_kind(kind: &str) -> bool { matches!(kind, "zone_thermostat" | "device_action" | "group_action") }
fn flow_action_kind(kind: &str) -> bool { matches!(kind, "zone_thermostat" | "device_action" | "group_action" | "ha_service_action") }
fn shared_input_comparison_kind(kind: &str) -> bool {
matches!(kind,
@@ -216,6 +216,44 @@ fn validate_condition(condition: &crate::models::FlowCondition, state: &AppState
let end = chrono::NaiveDate::parse_from_str(&end, "%Y-%m-%d").map_err(|_| AppError::BadRequest("invalid date range".into()))?;
if start > end { return Err(AppError::BadRequest("date range start must not be after end".into())); }
}
"cron_trigger" => {
let expr = flow_string(&condition.config, "expression").ok_or_else(|| AppError::BadRequest("cron block needs an expression".into()))?;
if !engine::cron_expression_valid(&expr) { return Err(AppError::BadRequest("invalid cron expression; expected 5 fields with *, */N, ranges or lists".into())); }
}
"stable_for" => {
let seconds = condition.config.get("seconds").and_then(Value::as_u64).unwrap_or(0);
if seconds == 0 || seconds > 604800 { return Err(AppError::BadRequest("stable-for duration must be between 1 second and 7 days".into())); }
if condition.inputs.len() != 1 { return Err(AppError::BadRequest("stable-for block needs exactly one input".into())); }
}
"delay" => {
let seconds = condition.config.get("seconds").and_then(Value::as_u64).unwrap_or(0);
if seconds == 0 || seconds > 604800 { return Err(AppError::BadRequest("delay must be between 1 second and 7 days".into())); }
if condition.inputs.len() != 1 { return Err(AppError::BadRequest("delay block needs exactly one input".into())); }
}
"rolling_stat" => {
let source = flow_string(&condition.config, "source").unwrap_or_default();
if !matches!(source.as_str(), "outdoor_temperature" | "device_temperature" | "zone_temperature" | "ha_numeric") { return Err(AppError::BadRequest("rolling statistic has an unsupported source".into())); }
if source == "device_temperature" { let id=flow_string(&condition.config,"device_id").unwrap_or_default(); if state.db.get_device(&id)?.is_none(){return Err(AppError::BadRequest("rolling statistic references a missing device".into()));} }
if source == "zone_temperature" { let id=flow_string(&condition.config,"zone_id").unwrap_or_default(); if state.db.get_zone(&id)?.is_none(){return Err(AppError::BadRequest("rolling statistic references a missing zone".into()));} }
if source == "ha_numeric" && flow_string(&condition.config,"entity_id").is_none() { return Err(AppError::BadRequest("rolling HA statistic needs entity_id".into())); }
let window = condition.config.get("window_seconds").and_then(Value::as_u64).unwrap_or(0);
if window < 10 || window > 604800 { return Err(AppError::BadRequest("rolling statistic window must be between 10 seconds and 7 days".into())); }
if !matches!(flow_string(&condition.config,"statistic").as_deref(), Some("mean") | Some("median")) { return Err(AppError::BadRequest("rolling statistic must be mean or median".into())); }
validate_flow_comparison(&condition.config)?;
}
"oscillates" => {
let source = flow_string(&condition.config, "source").unwrap_or_default();
if !matches!(source.as_str(), "outdoor_temperature" | "device_temperature" | "zone_temperature" | "ha_numeric") { return Err(AppError::BadRequest("oscillation block has an unsupported source".into())); }
if source == "device_temperature" { let id=flow_string(&condition.config,"device_id").unwrap_or_default(); if state.db.get_device(&id)?.is_none(){return Err(AppError::BadRequest("oscillation block references a missing device".into()));} }
if source == "zone_temperature" { let id=flow_string(&condition.config,"zone_id").unwrap_or_default(); if state.db.get_zone(&id)?.is_none(){return Err(AppError::BadRequest("oscillation block references a missing zone".into()));} }
if source == "ha_numeric" && flow_string(&condition.config,"entity_id").is_none() { return Err(AppError::BadRequest("oscillation HA source needs entity_id".into())); }
let window = condition.config.get("window_seconds").and_then(Value::as_u64).unwrap_or(0);
if window < 10 || window > 604800 { return Err(AppError::BadRequest("oscillation window must be between 10 seconds and 7 days".into())); }
let min_span = flow_f64(&condition.config, "min_span").unwrap_or(0.0);
if !min_span.is_finite() || min_span <= 0.0 { return Err(AppError::BadRequest("oscillation minimum span must be greater than zero".into())); }
let min_changes = condition.config.get("min_direction_changes").and_then(Value::as_u64).unwrap_or(0);
if min_changes == 0 || min_changes > 1000 { return Err(AppError::BadRequest("oscillation direction changes must be between 1 and 1000".into())); }
}
"outdoor_temperature" => { validate_flow_comparison(&condition.config)?; }
"device_temperature" => {
validate_flow_comparison(&condition.config)?;
@@ -364,12 +402,26 @@ fn compile_flow(state: &AppState, mut flow: crate::models::Flow) -> Result<(crat
let id = format!("flow:{}:automation:{}", flow.id, action_node.id);
let created_at = previous_automations.get(&id).map(|item| item.created_at.clone()).unwrap_or_else(|| now.clone());
let last_fired_at = previous_automations.get(&id).and_then(|item| item.last_fired_at.clone());
let flow_runtime = previous_automations.get(&id).map(|item| {
let previous_conditions = item.flow_conditions.iter().map(|condition| (condition.id.as_str(), condition)).collect::<std::collections::HashMap<_, _>>();
let current_conditions = conditions.iter().map(|condition| (condition.id.as_str(), condition)).collect::<std::collections::HashMap<_, _>>();
let mut runtime = item.flow_runtime.clone();
runtime.retain(|node_id, _| {
let Some(previous) = previous_conditions.get(node_id.as_str()) else { return false; };
let Some(current) = current_conditions.get(node_id.as_str()) else { return false; };
matches!(current.kind.as_str(), "stable_for" | "delay" | "rolling_stat" | "oscillates")
&& previous.kind == current.kind
&& previous.config == current.config
&& previous.inputs == current.inputs
});
runtime
}).unwrap_or_default();
let mut item = Automation {
id, name: generated_flow_name(&flow.id, &action_node.id), enabled: flow.enabled,
trigger_kind: "flow".into(), trigger_device_id: None, threshold: None, at_time: None,
action_device_id: String::new(), action_group_id: None, action_preset: None, action: DeviceCommand::default(), cooldown_seconds: 60,
last_fired_at, action_zone_id: None, action_zone_preset: None, flow_conditions: conditions,
flow_id: Some(flow.id.clone()), flow_node_id: Some(action_node.id.clone()), created_at, updated_at: now.clone(),
last_fired_at, action_zone_id: None, action_zone_preset: None, action_ha_domain: None, action_ha_service: None, action_ha_entity_id: None, action_ha_data: Value::Null, flow_conditions: conditions,
flow_id: Some(flow.id.clone()), flow_node_id: Some(action_node.id.clone()), flow_runtime, created_at, updated_at: now.clone(),
};
if let Some(cooldown) = action_node.config.get("cooldown_seconds").and_then(Value::as_u64) { item.cooldown_seconds = cooldown.max(30); }
match action_node.kind.as_str() {
@@ -403,6 +455,15 @@ fn compile_flow(state: &AppState, mut flow: crate::models::Flow) -> Result<(crat
item.action.sleep = flow_bool(&action_node.config, "sleep");
engine::validate_command(&item.action)?; if item.action.is_empty() { return Err(AppError::BadRequest("device action cannot be empty".into())); }
}
"ha_service_action" => {
let domain = flow_string(&action_node.config, "domain").ok_or_else(|| AppError::BadRequest("Home Assistant action needs a domain".into()))?;
let service = flow_string(&action_node.config, "service").ok_or_else(|| AppError::BadRequest("Home Assistant action needs a service".into()))?;
item.action_ha_domain = Some(domain);
item.action_ha_service = Some(service);
item.action_ha_entity_id = flow_string(&action_node.config, "entity_id");
item.action_ha_data = action_node.config.get("data").cloned().unwrap_or_else(|| json!({}));
if !item.action_ha_data.is_object() { return Err(AppError::BadRequest("Home Assistant service data must be a JSON object".into())); }
}
"group_action" => {
let id = flow_string(&action_node.config, "group_id").ok_or_else(|| AppError::BadRequest("group action needs a group".into()))?;
if !groups.iter().any(|g| g.id == id) { return Err(AppError::BadRequest("group action references a missing group".into())); }
@@ -581,6 +642,7 @@ fn dry_run_block_reason(state: &AppState, action: &crate::models::FlowNode) -> R
if climate_change && !resulting_enabled { return Ok(Some("group_control_disabled".into())); }
Ok(None)
}
"ha_service_action" => Ok(None),
_ => Ok(Some("unsupported_action".into())),
}
}
@@ -602,7 +664,9 @@ async fn simulate_flow(State(state): State<AppState>, Json(input): Json<FlowSimu
let mut actions = Vec::new();
for action in input.flow.nodes.iter().filter(|node| flow_action_kind(&node.kind)) {
let program = compile_flow_program(&action.id, &input.flow.nodes, &input.flow.edges)?;
let (matched, trace) = engine::evaluate_flow_conditions_trace(&state, &devices, &program, at.clone(), &input.overrides).await?;
let automation_id = format!("flow:{preview_id}:automation:{}", action.id);
let mut runtime = state.db.get_automation(&automation_id)?.map(|item| item.flow_runtime).unwrap_or_default();
let (matched, trace) = engine::evaluate_flow_conditions_trace(&state, &devices, &program, at.clone(), &input.overrides, Some(&mut runtime)).await?;
let blocked_reason = if matched && !input.flow.enabled { Some("flow_disabled".into()) }
else if matched { dry_run_block_reason(&state, action)? } else { None };
actions.push(json!({
src/engine/automations.rs
+216 -6
View File
@@ -28,12 +28,17 @@ async fn run_automations(state: &AppState) -> Result<()> {
"temperature_below" => find_temperature(&devices, item.trigger_device_id.as_deref())
.zip(item.threshold).map(|(t, threshold)| t < threshold).unwrap_or(false),
"time" => time_automation_due(&item, Local::now()),
"flow" => match flow_conditions_match(state, &devices, &item.flow_conditions).await {
"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
}
}
},
_ => false,
};
@@ -92,7 +97,9 @@ async fn run_automations(state: &AppState) -> Result<()> {
continue;
}
let target_devices: Vec<String> = if let Some(zone_id) = item.action_zone_id.as_deref() {
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)
@@ -115,7 +122,13 @@ async fn run_automations(state: &AppState) -> Result<()> {
continue;
}
let result: Result<bool, AppError> = if let Some(zone_id) = item.action_zone_id.as_deref() {
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() });
@@ -275,6 +288,108 @@ fn flow_zone_state_value(zone: &Zone, field: &str) -> Option<Value> {
})
}
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_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],
@@ -339,6 +454,10 @@ async fn flow_leaf_observation(
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);
@@ -468,6 +587,7 @@ pub async fn evaluate_flow_conditions_trace(
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();
@@ -502,6 +622,88 @@ pub async fn evaluate_flow_conditions_trace(
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}))
}
"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?;
@@ -524,14 +726,22 @@ pub async fn evaluate_flow_conditions_trace(
Ok((final_id.and_then(|id| values.get(&id).copied()).unwrap_or(false), trace))
}
async fn flow_conditions_match(state: &AppState, devices: &[Device], conditions: &[crate::models::FlowCondition]) -> Result<bool, AppError> {
async fn flow_conditions_match(state: &AppState, devices: &[Device], item: &mut Automation) -> Result<bool, AppError> {
let overrides = HashMap::new();
evaluate_flow_conditions_trace(state, devices, conditions, Local::now(), &overrides).await.map(|(matched, _)| matched)
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" | "outdoor_temperature" | "device_temperature" |
"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"
)
src/engine/tests.rs
+58 -1
View File
@@ -86,6 +86,63 @@ mod tests {
}
}
#[test]
fn flow_cron_validation_and_matching_cover_ranges_steps_and_sunday_alias() {
assert!(cron_expression_valid("*/5 * * * *"));
assert!(cron_expression_valid("0,15,30,45 6-18 * * 1-5"));
for invalid in ["* * * *", "*/0 * * * *", "61 * * * *", "* 24 * * *", "* * 0 * *", "* * * 13 *", "* * * * 8", "*/100 * * * *"] {
assert!(!cron_expression_valid(invalid), "accepted invalid cron: {invalid}");
}
let monday = Local.with_ymd_and_hms(2026, 9, 7, 10, 15, 0).single().unwrap();
assert!(cron_matches("15 10 * * 1", &monday));
assert!(cron_matches("15 10 * * 1-7", &monday));
assert!(!cron_matches("16 10 * * 1", &monday));
let sunday = Local.with_ymd_and_hms(2026, 9, 6, 8, 0, 0).single().unwrap();
assert!(cron_matches("0 8 * * 0", &sunday));
assert!(cron_matches("0 8 * * 7", &sunday));
assert!(cron_matches("0 8 * * 1-7", &sunday));
}
#[test]
fn flow_timed_gate_requires_continuity_and_resets_on_false() {
let mut state = crate::models::FlowRuntimeNodeState::default();
let start = Utc.with_ymd_and_hms(2026, 9, 2, 12, 0, 0).unwrap();
assert!(!flow_timed_gate_update(&mut state, true, 30, start));
assert!(!flow_timed_gate_update(&mut state, true, 30, start + chrono::Duration::seconds(29)));
assert!(flow_timed_gate_update(&mut state, true, 30, start + chrono::Duration::seconds(30)));
assert!(!flow_timed_gate_update(&mut state, false, 30, start + chrono::Duration::seconds(31)));
assert!(state.since.is_none());
assert!(!flow_timed_gate_update(&mut state, true, 30, start + chrono::Duration::seconds(32)));
}
#[test]
fn flow_rolling_statistics_compute_mean_median_and_prune_window() {
let mut state = crate::models::FlowRuntimeNodeState::default();
let start = Utc.with_ymd_and_hms(2026, 9, 2, 12, 0, 0).unwrap();
assert_eq!(flow_rolling_stat_update(&mut state, 10.0, 60, "mean", start), Some(10.0));
assert_eq!(flow_rolling_stat_update(&mut state, 20.0, 60, "mean", start + chrono::Duration::seconds(10)), Some(15.0));
assert_eq!(flow_rolling_stat_update(&mut state, 30.0, 60, "median", start + chrono::Duration::seconds(20)), Some(20.0));
assert_eq!(flow_rolling_stat_update(&mut state, 40.0, 60, "median", start + chrono::Duration::seconds(30)), Some(25.0));
assert_eq!(flow_rolling_stat_update(&mut state, 100.0, 60, "mean", start + chrono::Duration::seconds(100)), Some(100.0));
assert_eq!(state.samples.len(), 1);
}
#[test]
fn flow_oscillation_requires_direction_changes_not_only_large_span() {
let start = Utc.with_ymd_and_hms(2026, 9, 2, 12, 0, 0).unwrap();
let bouncing = [20.0, 22.0, 19.0, 23.0, 20.0].into_iter().enumerate().map(|(index, value)| crate::models::FlowRuntimeSample {
at: start + chrono::Duration::seconds(index as i64 * 10), value,
}).collect::<Vec<_>>();
let monotonic = [20.0, 21.0, 22.0, 23.0, 24.0].into_iter().enumerate().map(|(index, value)| crate::models::FlowRuntimeSample {
at: start + chrono::Duration::seconds(index as i64 * 10), value,
}).collect::<Vec<_>>();
let (span, changes) = flow_oscillation_metrics(&bouncing).unwrap();
assert!(span >= 4.0);
assert!(changes >= 2);
let (_, monotonic_changes) = flow_oscillation_metrics(&monotonic).unwrap();
assert_eq!(monotonic_changes, 0);
}
#[test]
fn time_automation_fires_only_once_in_the_same_minute() {
let now = Local.with_ymd_and_hms(2025, 1, 6, 10, 15, 40).single().unwrap();
@@ -94,7 +151,7 @@ mod tests {
trigger_device_id: None, threshold: None, at_time: Some("10:15".into()),
action_device_id: "d".into(), action_group_id: None, action_preset: None,
action: DeviceCommand { power: Some(true), ..Default::default() }, cooldown_seconds: 30,
last_fired_at: None, action_zone_id: None, action_zone_preset: None, flow_conditions: vec![], flow_id: None, flow_node_id: None, created_at: Utc::now(), updated_at: Utc::now(),
last_fired_at: None, action_zone_id: None, action_zone_preset: None, action_ha_domain: None, action_ha_service: None, action_ha_entity_id: None, action_ha_data: Value::Null, flow_conditions: vec![], flow_id: None, flow_node_id: None, flow_runtime: Default::default(), created_at: Utc::now(), updated_at: Utc::now(),
};
assert!(time_automation_due(&item, now.clone()));
item.last_fired_at = Some((now.clone() - chrono::Duration::seconds(35)).with_timezone(&Utc));
src/home_assistant.rs
+35
View File
@@ -145,3 +145,38 @@ pub async fn read_state(
let payload = read_entity(default_client, settings, entity_override).await?;
payload.get("state").and_then(Value::as_str).map(str::to_string).ok_or_else(|| anyhow!("Home Assistant state is missing"))
}
/// Call a Home Assistant service from a Flow action. The domain/service pair is explicit
/// and the payload is sent as JSON. Entity targeting is normalized through entity_id.
pub async fn call_service(
default_client: &reqwest::Client,
settings: &HomeAssistantSettings,
domain: &str,
service: &str,
entity_id: Option<&str>,
data: &Value,
) -> Result<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") }
if domain.trim().is_empty() || service.trim().is_empty() { bail!("Home Assistant domain/service is required") }
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(&format!("api/services/{}/{}", domain.trim(), service.trim())).context("cannot build Home Assistant service URL")?;
let mut payload = data.as_object().cloned().unwrap_or_default();
if let Some(entity) = entity_id.map(str::trim).filter(|v| !v.is_empty()) {
payload.insert("entity_id".into(), Value::String(entity.to_string()));
}
let client = request_client(default_client, settings)?;
let response = client.post(base).bearer_auth(settings.token.trim()).header("Accept", "application/json")
.json(&Value::Object(payload)).send().await.context("Home Assistant service request failed")?;
if !response.status().is_success() {
let status = response.status();
let body = response.text().await.unwrap_or_default();
bail!("Home Assistant returned {status}: {}", body.chars().take(200).collect::<String>())
}
match response.json::<Value>().await {
Ok(value) => Ok(value),
Err(_) => Ok(Value::Null),
}
}
src/models/automation.rs
+26
View File
@@ -25,6 +25,21 @@ pub struct Schedule {
pub flow_node_id: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
pub struct FlowRuntimeNodeState {
#[serde(default)]
pub since: Option<DateTime<Utc>>,
#[serde(default)]
pub samples: Vec<FlowRuntimeSample>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct FlowRuntimeSample {
pub at: DateTime<Utc>,
pub value: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Automation {
pub id: String,
@@ -59,6 +74,14 @@ pub struct Automation {
pub action_zone_id: Option<String>,
#[serde(default)]
pub action_zone_preset: Option<String>,
#[serde(default)]
pub action_ha_domain: Option<String>,
#[serde(default)]
pub action_ha_service: Option<String>,
#[serde(default)]
pub action_ha_entity_id: Option<String>,
#[serde(default)]
pub action_ha_data: Value,
/// Topologically ordered condition/logic program compiled from the visual Flow graph.
#[serde(default)]
pub flow_conditions: Vec<FlowCondition>,
@@ -66,6 +89,9 @@ pub struct Automation {
pub flow_id: Option<String>,
#[serde(default)]
pub flow_node_id: Option<String>,
/// Durable state for stateful Flow blocks (stable-for and rolling statistics).
#[serde(default)]
pub flow_runtime: BTreeMap<String, FlowRuntimeNodeState>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}