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

v0.11.1

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-09-02 22:58:55 +02:00
parent db6d2f09db
commit 563523d2e2
18 changed files with 420 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/flows.rs
+46 -2
View File
@@ -43,7 +43,7 @@ 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" | "cron_trigger" | "stable_for" | "delay" | "rolling_stat" | "oscillates" | "outdoor_temperature" | "device_temperature" |
"weekday" | "time_range" | "date_range" | "cron_trigger" | "stable_for" | "delay" | "state_duration" | "on_change" | "rate_limit" | "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"
)
@@ -136,6 +136,24 @@ fn validate_flow_graph(input: &FlowInput) -> Result<(), AppError> {
return Err(AppError::BadRequest("Flow connection has an unsupported target".into()));
}
}
for node in input.nodes.iter().filter(|node| node.kind == "rate_limit") {
let targets = input.edges.iter().filter(|edge| edge.from == node.id)
.filter_map(|edge| input.nodes.iter().find(|target| target.id == edge.to))
.collect::<Vec<_>>();
if targets.is_empty() || targets.iter().any(|target| !flow_action_kind(&target.kind)) {
return Err(AppError::BadRequest("rate-limit block must be placed directly before an action".into()));
}
}
for node in input.nodes.iter().filter(|node| node.kind == "on_change" && flow_string(&node.config, "mode").as_deref() == Some("value")) {
let sources = input.edges.iter().filter(|edge| edge.to == node.id)
.filter_map(|edge| input.nodes.iter().find(|source| source.id == edge.from))
.collect::<Vec<_>>();
if sources.len() != 1 || sources.iter().any(|source| flow_logic_kind(&source.kind) || matches!(source.kind.as_str(), "stable_for" | "delay" | "state_duration" | "on_change" | "rate_limit" | "rolling_stat" | "oscillates")) {
return Err(AppError::BadRequest("on-change value mode needs one direct source/condition input".into()));
}
}
// Reject cycles. Flow is deliberately a DAG: finite evaluation, deterministic topological order,
// and no hidden state machine semantics unless a dedicated stateful block is introduced later.
let mut outgoing = std::collections::HashMap::<String, Vec<String>>::new();
@@ -230,6 +248,32 @@ fn validate_condition(condition: &crate::models::FlowCondition, state: &AppState
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())); }
}
"state_duration" => {
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);
if min_seconds > 604800 || max_seconds.is_some_and(|value| value > 604800) {
return Err(AppError::BadRequest("state duration must be between 0 seconds and 7 days".into()));
}
if max_seconds.is_some_and(|value| value < min_seconds) {
return Err(AppError::BadRequest("state duration maximum must be greater than or equal to minimum".into()));
}
if min_seconds == 0 && max_seconds.is_none() {
return Err(AppError::BadRequest("state duration needs a minimum or maximum duration".into()));
}
if condition.inputs.len() != 1 { return Err(AppError::BadRequest("state duration block needs exactly one input".into())); }
}
"on_change" => {
let mode = flow_string(&condition.config, "mode").unwrap_or_else(|| "result".into());
if !matches!(mode.as_str(), "result" | "value") { return Err(AppError::BadRequest("on-change mode must be result or value".into())); }
if condition.inputs.len() != 1 { return Err(AppError::BadRequest("on-change block needs exactly one input".into())); }
}
"rate_limit" => {
let max_count = condition.config.get("max_count").and_then(Value::as_u64).unwrap_or(0);
let period_seconds = condition.config.get("period_seconds").and_then(Value::as_u64).unwrap_or(0);
if !(1..=1000).contains(&max_count) { return Err(AppError::BadRequest("rate limit max_count must be between 1 and 1000".into())); }
if !(1..=2678400).contains(&period_seconds) { return Err(AppError::BadRequest("rate limit period must be between 1 second and 31 days".into())); }
if condition.inputs.len() != 1 { return Err(AppError::BadRequest("rate-limit 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())); }
@@ -409,7 +453,7 @@ fn compile_flow(state: &AppState, mut flow: crate::models::Flow) -> Result<(crat
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")
matches!(current.kind.as_str(), "stable_for" | "delay" | "state_duration" | "on_change" | "rate_limit" | "rolling_stat" | "oscillates")
&& previous.kind == current.kind
&& previous.config == current.config
&& previous.inputs == current.inputs
src/engine/automations.rs
+105 -8
View File
@@ -21,13 +21,11 @@ async fn run_automations(state: &AppState) -> Result<()> {
let mut claimed_devices = std::collections::HashSet::<String>::new();
for mut item in automations {
if !item.enabled || !automation_ready(&item) { continue; }
if !item.enabled { continue; }
let ready = automation_ready(&item);
let should_fire = match item.trigger_kind.as_str() {
"temperature_above" => find_temperature(&devices, item.trigger_device_id.as_deref())
.zip(item.threshold).map(|(t, threshold)| t > threshold).unwrap_or(false),
"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()),
// 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;
@@ -40,9 +38,14 @@ async fn run_automations(state: &AppState) -> Result<()> {
}
}
},
"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 !should_fire { continue; }
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
@@ -153,7 +156,9 @@ async fn run_automations(state: &AppState) -> Result<()> {
match result {
Ok(true) => {
for device_id in target_devices { claimed_devices.insert(device_id); }
item.last_fired_at = Some(Utc::now());
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)?;
state.log("info", "automation.fired", &format!("Automation {} fired", item.name), json!({
@@ -369,6 +374,53 @@ fn flow_timed_gate_update(
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 != &current);
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,
@@ -606,6 +658,7 @@ pub async fn evaluate_flow_conditions_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)); }
@@ -634,6 +687,49 @@ pub async fn evaluate_flow_conditions_trace(
} 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("");
@@ -712,6 +808,7 @@ pub async fn evaluate_flow_conditions_trace(
_ => (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!({
src/engine/tests.rs
+31
View File
@@ -115,6 +115,37 @@ mod tests {
assert!(!flow_timed_gate_update(&mut state, true, 30, start + chrono::Duration::seconds(32)));
}
#[test]
fn flow_state_duration_respects_minimum_maximum_and_reset() {
let mut state = crate::models::FlowRuntimeNodeState::default();
let start = Utc.with_ymd_and_hms(2026, 9, 2, 12, 0, 0).unwrap();
assert_eq!(flow_state_duration_update(&mut state, true, 10, Some(30), start), (false, 0));
assert_eq!(flow_state_duration_update(&mut state, true, 10, Some(30), start + chrono::Duration::seconds(10)), (true, 10));
assert_eq!(flow_state_duration_update(&mut state, true, 10, Some(30), start + chrono::Duration::seconds(30)), (true, 30));
assert_eq!(flow_state_duration_update(&mut state, true, 10, Some(30), start + chrono::Duration::seconds(31)), (false, 31));
assert_eq!(flow_state_duration_update(&mut state, false, 10, Some(30), start + chrono::Duration::seconds(32)), (false, 0));
assert!(state.since.is_none());
}
#[test]
fn flow_change_gate_ignores_first_observation_and_detects_later_change() {
let mut state = crate::models::FlowRuntimeNodeState::default();
assert!(!flow_change_gate_update(&mut state, json!("off")));
assert!(!flow_change_gate_update(&mut state, json!("off")));
assert!(flow_change_gate_update(&mut state, json!("on")));
assert!(!flow_change_gate_update(&mut state, json!("on")));
}
#[test]
fn flow_rate_limit_prunes_rolling_window_and_blocks_at_limit() {
let mut state = crate::models::FlowRuntimeNodeState::default();
let start = Utc.with_ymd_and_hms(2026, 9, 2, 12, 0, 0).unwrap();
state.samples.push(crate::models::FlowRuntimeSample { at: start, value: 1.0 });
state.samples.push(crate::models::FlowRuntimeSample { at: start + chrono::Duration::seconds(10), value: 1.0 });
assert_eq!(flow_rate_limit_status(&mut state, 2, 60, start + chrono::Duration::seconds(20)), (false, 2));
assert_eq!(flow_rate_limit_status(&mut state, 2, 60, start + chrono::Duration::seconds(61)), (true, 1));
}
#[test]
fn flow_rolling_statistics_compute_mean_median_and_prune_window() {
let mut state = crate::models::FlowRuntimeNodeState::default();
src/models/automation.rs
+4 -1
View File
@@ -32,6 +32,9 @@ pub struct FlowRuntimeNodeState {
pub since: Option<DateTime<Utc>>,
#[serde(default)]
pub samples: Vec<FlowRuntimeSample>,
/// Last observed raw/result value for edge/change detection blocks.
#[serde(default)]
pub last_value: Option<Value>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -89,7 +92,7 @@ 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).
/// Durable state for stateful Flow blocks (timers, change detection, rate limits and rolling samples).
#[serde(default)]
pub flow_runtime: BTreeMap<String, FlowRuntimeNodeState>,
pub created_at: DateTime<Utc>,