v0.9.6
This commit is contained in:
Mateusz Gruszczyński
@@ -45,6 +45,7 @@ const SPA_ROUTES: &[&str] = &[
|
||||
"/groups",
|
||||
"/schedules",
|
||||
"/automations",
|
||||
"/flows",
|
||||
"/simulation",
|
||||
"/night-mode",
|
||||
"/home-assistant",
|
||||
@@ -84,6 +85,12 @@ pub fn router(state: AppState) -> Router {
|
||||
.route("/api/schedules/:id", get(get_schedule).put(update_schedule).delete(delete_schedule))
|
||||
.route("/api/automations", get(list_automations).post(create_automation))
|
||||
.route("/api/automations/:id", get(get_automation).put(update_automation).delete(delete_automation))
|
||||
.route("/api/flows", get(list_flows).post(create_flow))
|
||||
.route("/api/flows/import", post(import_flow))
|
||||
.route("/api/flows/simulate", post(simulate_flow))
|
||||
.route("/api/flows/:id/export", get(export_flow))
|
||||
.route("/api/flows/:id/logs", get(flow_logs))
|
||||
.route("/api/flows/:id", get(get_flow).put(update_flow).delete(delete_flow))
|
||||
.route("/api/readings", get(readings))
|
||||
.route("/api/history", get(history))
|
||||
.route("/api/control-plan", get(control_plan))
|
||||
@@ -125,6 +132,7 @@ pub fn router(state: AppState) -> Router {
|
||||
.route("/manifest.webmanifest", get(manifest))
|
||||
.route("/sw.js", get(service_worker))
|
||||
.route("/favicon.svg", get(favicon))
|
||||
.route("/flows/:id", get(index))
|
||||
.route("/lang/index.json", get(language_index))
|
||||
.route("/lang/:file", get(language_file))
|
||||
.merge(protected)
|
||||
@@ -167,6 +175,7 @@ include!("api/groups.rs");
|
||||
include!("api/house.rs");
|
||||
include!("api/schedules.rs");
|
||||
include!("api/automations.rs");
|
||||
include!("api/flows.rs");
|
||||
include!("api/history.rs");
|
||||
include!("api/events.rs");
|
||||
include!("api/settings.rs");
|
||||
|
||||
@@ -87,6 +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,
|
||||
created_at, updated_at: Utc::now() }
|
||||
}
|
||||
}
|
||||
@@ -140,6 +141,7 @@ async fn update_automation(State(state): State<AppState>, Path(id): Path<String>
|
||||
let _automation_guard = state.lock_automation_operation().await;
|
||||
input.validate()?;
|
||||
let existing = state.db.get_automation(&id)?.ok_or_else(|| AppError::NotFound(format!("automation {id}")))?;
|
||||
if existing.flow_id.is_some() { return Err(AppError::BadRequest("this automation is generated by Flow; edit it in the Flow editor".into())); }
|
||||
let action_group_id = input.action_group_id.as_deref().map(str::trim).filter(|value| !value.is_empty()).map(str::to_string);
|
||||
let _group_guard = if let Some(group_id) = action_group_id.as_deref() {
|
||||
Some(state.lock_group_operation(group_id).await)
|
||||
@@ -153,6 +155,8 @@ async fn update_automation(State(state): State<AppState>, Path(id): Path<String>
|
||||
async fn delete_automation(State(state): State<AppState>, Path(id): Path<String>) -> Result<StatusCode, AppError> {
|
||||
let _configuration_guard = state.lock_configuration_operation().await;
|
||||
let _automation_guard = state.lock_automation_operation().await;
|
||||
let existing = state.db.get_automation(&id)?.ok_or_else(|| AppError::NotFound(format!("automation {id}")))?;
|
||||
if existing.flow_id.is_some() { return Err(AppError::BadRequest("this automation is generated by Flow; delete it from the Flow editor".into())); }
|
||||
if !state.db.delete_automation(&id)? { return Err(AppError::NotFound(format!("automation {id}"))); }
|
||||
state.broadcast("automation.deleted", json!({"id": id}));
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
|
||||
@@ -0,0 +1,561 @@
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct FlowInput {
|
||||
name: String,
|
||||
#[serde(default = "yes")]
|
||||
enabled: bool,
|
||||
#[serde(default)]
|
||||
description: String,
|
||||
#[serde(default)]
|
||||
nodes: Vec<crate::models::FlowNode>,
|
||||
#[serde(default)]
|
||||
edges: Vec<crate::models::FlowEdge>,
|
||||
#[serde(default)]
|
||||
expected_revision: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct FlowSimulationInput {
|
||||
flow: FlowInput,
|
||||
#[serde(default)]
|
||||
flow_id: Option<String>,
|
||||
#[serde(default)]
|
||||
at: Option<String>,
|
||||
#[serde(default)]
|
||||
overrides: std::collections::HashMap<String, Value>,
|
||||
#[serde(default = "yes")]
|
||||
log: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct FlowLogsQuery { limit: Option<u32> }
|
||||
|
||||
fn flow_string(config: &Value, key: &str) -> Option<String> {
|
||||
config.get(key).and_then(Value::as_str).map(str::trim).filter(|v| !v.is_empty()).map(str::to_string)
|
||||
}
|
||||
fn flow_f64(config: &Value, key: &str) -> Option<f64> { config.get(key).and_then(Value::as_f64) }
|
||||
fn flow_u8(config: &Value, key: &str) -> Option<u8> { config.get(key).and_then(Value::as_u64).and_then(|value| u8::try_from(value).ok()) }
|
||||
fn flow_bool(config: &Value, key: &str) -> Option<bool> { config.get(key).and_then(Value::as_bool) }
|
||||
|
||||
fn flow_condition_kind(kind: &str) -> bool {
|
||||
matches!(kind,
|
||||
"weekday" | "time_range" | "date_range" | "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"
|
||||
)
|
||||
}
|
||||
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 validate_flow_graph(input: &FlowInput) -> Result<(), AppError> {
|
||||
if input.name.trim().is_empty() { return Err(AppError::BadRequest("flow name is required".into())); }
|
||||
if input.nodes.is_empty() { return Err(AppError::BadRequest("flow needs at least one block".into())); }
|
||||
let mut ids = std::collections::HashSet::<String>::new();
|
||||
for node in &input.nodes {
|
||||
if node.id.trim().is_empty() || !ids.insert(node.id.clone()) { return Err(AppError::BadRequest("flow contains duplicate or empty block IDs".into())); }
|
||||
if !(flow_condition_kind(&node.kind) || flow_action_kind(&node.kind) || flow_logic_kind(&node.kind)) {
|
||||
return Err(AppError::BadRequest(format!("unsupported flow block: {}", node.kind)));
|
||||
}
|
||||
}
|
||||
if !input.nodes.iter().any(|node| flow_action_kind(&node.kind)) { return Err(AppError::BadRequest("flow needs at least one action block".into())); }
|
||||
let mut edge_ids = std::collections::HashSet::<String>::new();
|
||||
let mut connections = std::collections::HashSet::<(String, String)>::new();
|
||||
for edge in &input.edges {
|
||||
if edge.id.trim().is_empty() || !edge_ids.insert(edge.id.clone()) || !connections.insert((edge.from.clone(), edge.to.clone())) {
|
||||
return Err(AppError::BadRequest("flow contains duplicate or empty connection IDs".into()));
|
||||
}
|
||||
if edge.from == edge.to || !ids.contains(edge.from.as_str()) || !ids.contains(edge.to.as_str()) {
|
||||
return Err(AppError::BadRequest("flow contains an invalid connection".into()));
|
||||
}
|
||||
let from = input.nodes.iter().find(|node| node.id == edge.from).expect("validated Flow source");
|
||||
let to = input.nodes.iter().find(|node| node.id == edge.to).expect("validated Flow target");
|
||||
if flow_action_kind(&from.kind) {
|
||||
return Err(AppError::BadRequest("Flow action blocks must be terminal and cannot feed another block".into()));
|
||||
}
|
||||
if !(flow_condition_kind(&to.kind) || flow_logic_kind(&to.kind) || flow_action_kind(&to.kind)) {
|
||||
return Err(AppError::BadRequest("Flow connection has an unsupported target".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();
|
||||
for edge in &input.edges { outgoing.entry(edge.from.clone()).or_default().push(edge.to.clone()); }
|
||||
fn visit(id: &str, outgoing: &std::collections::HashMap<String, Vec<String>>, temp: &mut std::collections::HashSet<String>, done: &mut std::collections::HashSet<String>) -> bool {
|
||||
if done.contains(id) { return false; }
|
||||
if !temp.insert(id.to_string()) { return true; }
|
||||
if outgoing.get(id).into_iter().flatten().any(|next| visit(next, outgoing, temp, done)) { return true; }
|
||||
temp.remove(id); done.insert(id.to_string()); false
|
||||
}
|
||||
let mut temp = std::collections::HashSet::new(); let mut done = std::collections::HashSet::new();
|
||||
for id in &ids { if visit(id, &outgoing, &mut temp, &mut done) { return Err(AppError::BadRequest("flow connections cannot contain a cycle".into())); } }
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn compile_flow_program(action_id: &str, nodes: &[crate::models::FlowNode], edges: &[crate::models::FlowEdge]) -> Result<Vec<crate::models::FlowCondition>, AppError> {
|
||||
let by_id: std::collections::HashMap<String, crate::models::FlowNode> = nodes.iter().cloned().map(|node| (node.id.clone(), node)).collect();
|
||||
let mut incoming = std::collections::HashMap::<String, Vec<String>>::new();
|
||||
for edge in edges { incoming.entry(edge.to.clone()).or_default().push(edge.from.clone()); }
|
||||
let action_inputs = incoming.get(action_id).cloned().unwrap_or_default();
|
||||
if action_inputs.is_empty() { return Err(AppError::BadRequest(format!("action '{action_id}' needs at least one connected condition"))); }
|
||||
|
||||
fn visit(
|
||||
id: &str,
|
||||
by_id: &std::collections::HashMap<String, crate::models::FlowNode>,
|
||||
incoming: &std::collections::HashMap<String, Vec<String>>,
|
||||
seen: &mut std::collections::HashSet<String>,
|
||||
out: &mut Vec<crate::models::FlowCondition>,
|
||||
) -> Result<(), AppError> {
|
||||
if !seen.insert(id.to_string()) { return Ok(()); }
|
||||
let node = by_id.get(id).ok_or_else(|| AppError::BadRequest("Flow references a missing block".into()))?;
|
||||
if !(flow_condition_kind(&node.kind) || flow_logic_kind(&node.kind)) {
|
||||
return Err(AppError::BadRequest("only condition or logic blocks can feed a Flow action".into()));
|
||||
}
|
||||
let inputs = incoming.get(id).cloned().unwrap_or_default();
|
||||
for input in &inputs { visit(input, by_id, incoming, seen, out)?; }
|
||||
match node.kind.as_str() {
|
||||
"logic_not" if inputs.len() != 1 => return Err(AppError::BadRequest("NOT block needs exactly one input".into())),
|
||||
"logic_and" | "logic_or" if inputs.is_empty() => return Err(AppError::BadRequest(format!("{} block needs at least one input", node.kind))),
|
||||
_ => {}
|
||||
}
|
||||
out.push(crate::models::FlowCondition { id: node.id.clone(), kind: node.kind.clone(), config: node.config.clone(), inputs });
|
||||
Ok(())
|
||||
}
|
||||
|
||||
let mut out = Vec::new();
|
||||
let mut seen = std::collections::HashSet::new();
|
||||
for input in &action_inputs { visit(input, &by_id, &incoming, &mut seen, &mut out)?; }
|
||||
out.push(crate::models::FlowCondition {
|
||||
id: format!("__flow_action__:{action_id}"),
|
||||
kind: "logic_and".into(),
|
||||
config: json!({}),
|
||||
inputs: action_inputs,
|
||||
});
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
fn validate_text_comparison(config: &Value) -> Result<(), AppError> {
|
||||
let op = flow_string(config, "operator").unwrap_or_else(|| "eq".into());
|
||||
if !matches!(op.as_str(), "eq" | "neq") { return Err(AppError::BadRequest("state operator must be eq or neq".into())); }
|
||||
if config.get("value").is_none() { return Err(AppError::BadRequest("state block needs a value".into())); }
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_condition(condition: &crate::models::FlowCondition, state: &AppState) -> Result<(), AppError> {
|
||||
match condition.kind.as_str() {
|
||||
"weekday" => {
|
||||
let days = condition.config.get("days").and_then(Value::as_array).ok_or_else(|| AppError::BadRequest("weekday block needs days".into()))?;
|
||||
if days.is_empty() || days.iter().any(|v| v.as_u64().map(|d| !(1..=7).contains(&d)).unwrap_or(true)) { return Err(AppError::BadRequest("weekday block contains invalid days".into())); }
|
||||
}
|
||||
"time_range" => {
|
||||
for key in ["start", "end"] { let value = flow_string(&condition.config, key).ok_or_else(|| AppError::BadRequest(format!("time range needs {key}")))?; chrono::NaiveTime::parse_from_str(&value, "%H:%M").map_err(|_| AppError::BadRequest("invalid time range".into()))?; }
|
||||
}
|
||||
"date_range" => {
|
||||
let start = flow_string(&condition.config, "start").ok_or_else(|| AppError::BadRequest("date range needs start".into()))?;
|
||||
let end = flow_string(&condition.config, "end").ok_or_else(|| AppError::BadRequest("date range needs end".into()))?;
|
||||
let start = chrono::NaiveDate::parse_from_str(&start, "%Y-%m-%d").map_err(|_| AppError::BadRequest("invalid date range".into()))?;
|
||||
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())); }
|
||||
}
|
||||
"outdoor_temperature" => { validate_flow_comparison(&condition.config)?; }
|
||||
"device_temperature" => {
|
||||
validate_flow_comparison(&condition.config)?;
|
||||
let id = flow_string(&condition.config, "device_id").ok_or_else(|| AppError::BadRequest("device temperature block needs device".into()))?;
|
||||
if state.db.get_device(&id)?.is_none() { return Err(AppError::BadRequest("flow references a missing device".into())); }
|
||||
}
|
||||
"zone_temperature" => {
|
||||
validate_flow_comparison(&condition.config)?;
|
||||
let id = flow_string(&condition.config, "zone_id").ok_or_else(|| AppError::BadRequest("zone temperature block needs zone".into()))?;
|
||||
if state.db.get_zone(&id)?.is_none() { return Err(AppError::BadRequest("flow references a missing zone".into())); }
|
||||
}
|
||||
"ha_state" => {
|
||||
if flow_string(&condition.config, "entity_id").is_none() { return Err(AppError::BadRequest("Home Assistant state block needs entity_id".into())); }
|
||||
validate_text_comparison(&condition.config)?;
|
||||
}
|
||||
"ha_numeric" => {
|
||||
if flow_string(&condition.config, "entity_id").is_none() { return Err(AppError::BadRequest("Home Assistant numeric block needs entity_id".into())); }
|
||||
validate_flow_comparison(&condition.config)?;
|
||||
}
|
||||
"ha_attribute" => {
|
||||
if flow_string(&condition.config, "entity_id").is_none() || flow_string(&condition.config, "attribute").is_none() {
|
||||
return Err(AppError::BadRequest("Home Assistant attribute block needs entity_id and attribute".into()));
|
||||
}
|
||||
let op = flow_string(&condition.config, "operator").unwrap_or_else(|| "eq".into());
|
||||
if !matches!(op.as_str(), "lt" | "lte" | "gt" | "gte" | "eq" | "neq") { return Err(AppError::BadRequest("unsupported Home Assistant attribute operator".into())); }
|
||||
if condition.config.get("value").is_none() { return Err(AppError::BadRequest("Home Assistant attribute block needs a value".into())); }
|
||||
}
|
||||
"ha_available" => {
|
||||
if flow_string(&condition.config, "entity_id").is_none() { return Err(AppError::BadRequest("Home Assistant availability block needs entity_id".into())); }
|
||||
}
|
||||
"house_mode" => {
|
||||
let value = flow_string(&condition.config, "value").ok_or_else(|| AppError::BadRequest("house mode block needs a mode".into()))?;
|
||||
if !matches!(value.as_str(), "cool" | "heat" | "off") { return Err(AppError::BadRequest("house mode must be cool, heat or off".into())); }
|
||||
validate_text_comparison(&condition.config)?;
|
||||
}
|
||||
"device_state" => {
|
||||
let id = flow_string(&condition.config, "device_id").ok_or_else(|| AppError::BadRequest("device state block needs device".into()))?;
|
||||
if state.db.get_device(&id)?.is_none() { return Err(AppError::BadRequest("flow references a missing device".into())); }
|
||||
let field = flow_string(&condition.config, "field").ok_or_else(|| AppError::BadRequest("device state block needs a field".into()))?;
|
||||
if !matches!(field.as_str(), "enabled" | "online" | "power" | "mode" | "fan_speed" | "swing_vertical" | "swing_horizontal" | "quiet" | "turbo" | "light" | "air" | "xfan" | "health" | "sleep") { return Err(AppError::BadRequest("unsupported device state field".into())); }
|
||||
validate_text_comparison(&condition.config)?;
|
||||
}
|
||||
"zone_state" => {
|
||||
let id = flow_string(&condition.config, "zone_id").ok_or_else(|| AppError::BadRequest("zone state block needs zone".into()))?;
|
||||
if state.db.get_zone(&id)?.is_none() { return Err(AppError::BadRequest("flow references a missing zone".into())); }
|
||||
let field = flow_string(&condition.config, "field").ok_or_else(|| AppError::BadRequest("zone state block needs a field".into()))?;
|
||||
if !matches!(field.as_str(), "enabled" | "mode" | "active_preset" | "demand" | "control_owner" | "device_manual_override" | "local_thermostat_power") { return Err(AppError::BadRequest("unsupported zone state field".into())); }
|
||||
validate_text_comparison(&condition.config)?;
|
||||
}
|
||||
"group_state" => {
|
||||
let id = flow_string(&condition.config, "group_id").ok_or_else(|| AppError::BadRequest("group state block needs group".into()))?;
|
||||
if state.db.get_group(&id)?.is_none() { return Err(AppError::BadRequest("flow references a missing group".into())); }
|
||||
let field = flow_string(&condition.config, "field").ok_or_else(|| AppError::BadRequest("group state block needs a field".into()))?;
|
||||
if field != "power_enabled" { return Err(AppError::BadRequest("unsupported group state field".into())); }
|
||||
validate_text_comparison(&condition.config)?;
|
||||
}
|
||||
"night_mode" => {}
|
||||
"constant" => {
|
||||
if condition.config.get("value").and_then(Value::as_bool).is_none() { return Err(AppError::BadRequest("constant block needs a boolean value".into())); }
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn validate_flow_comparison(config: &Value) -> Result<(), AppError> {
|
||||
let op = flow_string(config, "operator").unwrap_or_else(|| "lt".into());
|
||||
if !matches!(op.as_str(), "lt" | "lte" | "gt" | "gte" | "eq" | "neq") { return Err(AppError::BadRequest("unsupported comparison operator".into())); }
|
||||
if flow_f64(config, "value").is_none() { return Err(AppError::BadRequest("comparison block needs a numeric value".into())); }
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn compile_flow(state: &AppState, mut flow: crate::models::Flow) -> Result<(crate::models::Flow, Vec<Schedule>, Vec<Automation>), AppError> {
|
||||
let mut schedules = Vec::new();
|
||||
let mut automations = Vec::new();
|
||||
let now = Utc::now();
|
||||
let zones = state.db.list_zones()?;
|
||||
let groups = state.db.list_groups()?;
|
||||
let devices = state.db.list_devices()?;
|
||||
let previous_schedules: std::collections::HashMap<String, Schedule> = state.db.list_schedules()?.into_iter()
|
||||
.filter(|item| item.flow_id.as_deref() == Some(flow.id.as_str())).map(|item| (item.id.clone(), item)).collect();
|
||||
let previous_automations: std::collections::HashMap<String, Automation> = state.db.list_automations()?.into_iter()
|
||||
.filter(|item| item.flow_id.as_deref() == Some(flow.id.as_str())).map(|item| (item.id.clone(), item)).collect();
|
||||
let actions: Vec<_> = flow.nodes.iter().filter(|node| flow_action_kind(&node.kind)).cloned().collect();
|
||||
for action_node in actions {
|
||||
let conditions = compile_flow_program(&action_node.id, &flow.nodes, &flow.edges)?;
|
||||
for condition in conditions.iter().filter(|condition| flow_condition_kind(&condition.kind)) { validate_condition(condition, state)?; }
|
||||
let schedule_leaves: Vec<_> = conditions.iter().filter(|condition| flow_condition_kind(&condition.kind)).collect();
|
||||
let preset_for_schedule = flow_string(&action_node.config, "preset").unwrap_or_else(|| "comfort".into());
|
||||
let mode_for_schedule = flow_string(&action_node.config, "mode").unwrap_or_else(|| "auto".into());
|
||||
let native_time_range = schedule_leaves.iter().find(|condition| condition.kind == "time_range")
|
||||
.and_then(|condition| flow_string(&condition.config, "start").zip(flow_string(&condition.config, "end")))
|
||||
.and_then(|(start, end)| chrono::NaiveTime::parse_from_str(&start, "%H:%M").ok().zip(chrono::NaiveTime::parse_from_str(&end, "%H:%M").ok()))
|
||||
.map(|(start, end)| start < end)
|
||||
.unwrap_or(false);
|
||||
let schedule_only = action_node.kind == "zone_thermostat"
|
||||
&& !conditions.iter().any(|condition| matches!(condition.kind.as_str(), "logic_or" | "logic_not"))
|
||||
&& schedule_leaves.iter().all(|condition| matches!(condition.kind.as_str(), "weekday" | "time_range"))
|
||||
&& schedule_leaves.iter().filter(|condition| condition.kind == "weekday").count() == 1
|
||||
&& schedule_leaves.iter().filter(|condition| condition.kind == "time_range").count() == 1
|
||||
&& native_time_range
|
||||
&& preset_for_schedule != "auto"
|
||||
&& mode_for_schedule == "auto"
|
||||
&& flow_bool(&action_node.config, "power").is_none();
|
||||
if schedule_only {
|
||||
let zone_id = flow_string(&action_node.config, "zone_id").ok_or_else(|| AppError::BadRequest("thermostat block needs a zone".into()))?;
|
||||
let zone = zones.iter().find(|z| z.id == zone_id).ok_or_else(|| AppError::BadRequest("thermostat block references a missing zone".into()))?;
|
||||
let preset = flow_string(&action_node.config, "preset").unwrap_or_else(|| "comfort".into());
|
||||
if !matches!(preset.as_str(), "comfort" | "sleep" | "away" | "custom") { return Err(AppError::BadRequest("unsupported thermostat preset".into())); }
|
||||
let setpoint = flow_f64(&action_node.config, "setpoint").unwrap_or(zone.setpoint);
|
||||
if preset == "custom" && !(8.0..=30.0).contains(&setpoint) { return Err(AppError::BadRequest("custom thermostat target must be between 8 and 30 C".into())); }
|
||||
let weekday = schedule_leaves.iter().find(|c| c.kind == "weekday").copied().unwrap();
|
||||
let time = schedule_leaves.iter().find(|c| c.kind == "time_range").copied().unwrap();
|
||||
let weekdays = weekday.config.get("days").and_then(Value::as_array).unwrap().iter().filter_map(Value::as_u64).map(|v| v as u32).collect();
|
||||
let id = format!("flow:{}:schedule:{}", flow.id, action_node.id);
|
||||
let created_at = previous_schedules.get(&id).map(|item| item.created_at.clone()).unwrap_or_else(|| now.clone());
|
||||
let schedule = Schedule {
|
||||
id, zone_id, name: format!("{} · {}", flow.name, zone.name), enabled: flow.enabled,
|
||||
weekdays, start_time: flow_string(&time.config, "start").unwrap(), end_time: flow_string(&time.config, "end").unwrap(), preset, setpoint,
|
||||
created_at, updated_at: now.clone(), flow_id: Some(flow.id.clone()), flow_node_id: Some(action_node.id.clone()),
|
||||
};
|
||||
schedules.push(schedule);
|
||||
continue;
|
||||
}
|
||||
drop(schedule_leaves);
|
||||
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 mut item = Automation {
|
||||
id, name: format!("{} · {}", flow.name, 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(),
|
||||
};
|
||||
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() {
|
||||
"zone_thermostat" => {
|
||||
let zone_id = flow_string(&action_node.config, "zone_id").ok_or_else(|| AppError::BadRequest("thermostat block needs a zone".into()))?;
|
||||
if !zones.iter().any(|z| z.id == zone_id) { return Err(AppError::BadRequest("thermostat block references a missing zone".into())); }
|
||||
item.action_zone_id = Some(zone_id);
|
||||
let preset = flow_string(&action_node.config, "preset").unwrap_or_else(|| "comfort".into());
|
||||
if !matches!(preset.as_str(), "comfort" | "sleep" | "away" | "custom" | "auto") { return Err(AppError::BadRequest("unsupported thermostat preset".into())); }
|
||||
item.action_zone_preset = Some(preset.clone());
|
||||
if preset == "custom" { let target = flow_f64(&action_node.config, "setpoint").ok_or_else(|| AppError::BadRequest("custom thermostat block needs a target".into()))?; if !(8.0..=30.0).contains(&target) { return Err(AppError::BadRequest("custom thermostat target must be between 8 and 30 C".into())); } item.action.target_temperature = Some(target); }
|
||||
if let Some(power) = flow_bool(&action_node.config, "power") { item.action.power = Some(power); }
|
||||
if let Some(mode) = flow_string(&action_node.config, "mode") { if !matches!(mode.as_str(), "auto" | "heat" | "cool") { return Err(AppError::BadRequest("unsupported thermostat mode".into())); } item.action.mode = Some(mode); }
|
||||
}
|
||||
"device_action" => {
|
||||
let id = flow_string(&action_node.config, "device_id").ok_or_else(|| AppError::BadRequest("device action needs a device".into()))?;
|
||||
if !devices.iter().any(|d| d.id == id) { return Err(AppError::BadRequest("device action references a missing device".into())); }
|
||||
item.action_device_id = id;
|
||||
item.action.power = flow_bool(&action_node.config, "power");
|
||||
item.action.mode = flow_string(&action_node.config, "mode");
|
||||
item.action.target_temperature = flow_f64(&action_node.config, "target_temperature");
|
||||
item.action.fan_speed = flow_u8(&action_node.config, "fan_speed");
|
||||
item.action.swing_vertical = flow_bool(&action_node.config, "swing_vertical");
|
||||
item.action.swing_horizontal = flow_bool(&action_node.config, "swing_horizontal");
|
||||
item.action.quiet = flow_bool(&action_node.config, "quiet");
|
||||
item.action.turbo = flow_bool(&action_node.config, "turbo");
|
||||
item.action.light = flow_bool(&action_node.config, "light");
|
||||
item.action.air = flow_bool(&action_node.config, "air");
|
||||
item.action.xfan = flow_bool(&action_node.config, "xfan");
|
||||
item.action.health = flow_bool(&action_node.config, "health");
|
||||
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())); }
|
||||
}
|
||||
"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())); }
|
||||
item.action_group_id = Some(id); item.action.power = flow_bool(&action_node.config, "power"); item.action.mode = flow_string(&action_node.config, "mode"); item.action_preset = flow_string(&action_node.config, "preset");
|
||||
if let Some(mode) = item.action.mode.as_deref() {
|
||||
if !matches!(mode, "auto" | "house" | "cool" | "heat") { return Err(AppError::BadRequest("unsupported Flow group mode".into())); }
|
||||
}
|
||||
if let Some(preset) = item.action_preset.as_deref() {
|
||||
if !matches!(preset, "auto" | "comfort" | "sleep" | "away" | "custom") { return Err(AppError::BadRequest("unsupported Flow group preset".into())); }
|
||||
if preset == "custom" {
|
||||
let target = flow_f64(&action_node.config, "setpoint").ok_or_else(|| AppError::BadRequest("custom Flow group preset needs a target".into()))?;
|
||||
if !(8.0..=30.0).contains(&target) { return Err(AppError::BadRequest("custom Flow group target must be between 8 and 30 C".into())); }
|
||||
item.action.target_temperature = Some(target);
|
||||
}
|
||||
}
|
||||
if item.action.power.is_none() && item.action.mode.is_none() && item.action_preset.is_none() { return Err(AppError::BadRequest("group action cannot be empty".into())); }
|
||||
}
|
||||
_ => unreachable!(),
|
||||
}
|
||||
automations.push(item);
|
||||
}
|
||||
let existing: Vec<_> = state.db.list_schedules()?.into_iter().filter(|s| s.flow_id.as_deref() != Some(flow.id.as_str())).collect();
|
||||
for candidate in &schedules {
|
||||
for other in existing.iter().chain(schedules.iter().filter(|s| s.id != candidate.id)) {
|
||||
if engine::schedules_overlap(candidate, other) { return Err(AppError::BadRequest(format!("flow schedule '{}' overlaps with '{}'", candidate.name, other.name))); }
|
||||
}
|
||||
}
|
||||
flow.compiled_schedule_ids = schedules.iter().map(|s| s.id.clone()).collect();
|
||||
flow.compiled_automation_ids = automations.iter().map(|a| a.id.clone()).collect();
|
||||
flow.summary = format!("{} bloków · {} harmonogramów · {} automatyzacji", flow.nodes.len(), schedules.len(), automations.len());
|
||||
Ok((flow, schedules, automations))
|
||||
}
|
||||
|
||||
fn flow_from_input(id: String, input: FlowInput, created_at: chrono::DateTime<Utc>, revision: u64) -> crate::models::Flow {
|
||||
crate::models::Flow {
|
||||
id, name: input.name.trim().into(), enabled: input.enabled, description: input.description.trim().into(), nodes: input.nodes, edges: input.edges,
|
||||
summary: String::new(), compiled_schedule_ids: vec![], compiled_automation_ids: vec![], revision, created_at, updated_at: Utc::now(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn list_flows(State(state): State<AppState>) -> Result<Json<Vec<crate::models::Flow>>, AppError> { Ok(Json(state.db.list_flows()?)) }
|
||||
async fn get_flow(State(state): State<AppState>, Path(id): Path<String>) -> Result<Json<crate::models::Flow>, AppError> { state.db.get_flow(&id)?.map(Json).ok_or_else(|| AppError::NotFound(format!("flow {id}"))) }
|
||||
|
||||
async fn create_flow(State(state): State<AppState>, Json(input): Json<FlowInput>) -> Result<(StatusCode, Json<crate::models::Flow>), AppError> {
|
||||
validate_flow_graph(&input)?;
|
||||
let _configuration_guard = state.lock_configuration_operation().await;
|
||||
let _automation_guard = state.lock_automation_operation().await;
|
||||
let _schedule_guard = state.lock_schedule_operation().await;
|
||||
let _cycle_guard = state.lock_zone_control_cycle().await;
|
||||
let flow = flow_from_input(Uuid::new_v4().to_string(), input, Utc::now(), 1);
|
||||
let (flow, schedules, automations) = compile_flow(&state, flow)?;
|
||||
state.db.replace_flow_outputs(&flow, &schedules, &automations)?;
|
||||
for zone_id in schedules.iter().map(|s| s.zone_id.as_str()).collect::<std::collections::HashSet<_>>() { refresh_zone_override_boundary(&state, zone_id).await?; }
|
||||
state.log("info", "flow.created", &format!("Created Flow {}", flow.name), json!({"flow_id": flow.id, "revision": flow.revision, "schedules": schedules.len(), "automations": automations.len()}));
|
||||
state.broadcast("flow.created", serde_json::to_value(&flow)?); state.wake_zone_control();
|
||||
Ok((StatusCode::CREATED, Json(flow)))
|
||||
}
|
||||
|
||||
async fn update_flow(State(state): State<AppState>, Path(id): Path<String>, Json(input): Json<FlowInput>) -> Result<Json<crate::models::Flow>, AppError> {
|
||||
validate_flow_graph(&input)?;
|
||||
let _configuration_guard = state.lock_configuration_operation().await;
|
||||
let _automation_guard = state.lock_automation_operation().await;
|
||||
let _schedule_guard = state.lock_schedule_operation().await;
|
||||
let _cycle_guard = state.lock_zone_control_cycle().await;
|
||||
let existing = state.db.get_flow(&id)?.ok_or_else(|| AppError::NotFound(format!("flow {id}")))?;
|
||||
let expected = input.expected_revision.ok_or_else(|| AppError::BadRequest("Flow update requires expected_revision".into()))?;
|
||||
if expected != existing.revision { return Err(AppError::Conflict(format!("flow {id} changed; expected revision {expected}, current revision {}", existing.revision))); }
|
||||
let old_zone_ids: std::collections::HashSet<String> = state.db.list_schedules()?.into_iter().filter(|s| s.flow_id.as_deref() == Some(id.as_str())).map(|s| s.zone_id).collect();
|
||||
let next_revision = existing.revision.saturating_add(1).max(1);
|
||||
let flow = flow_from_input(id, input, existing.created_at, next_revision);
|
||||
let (flow, schedules, automations) = compile_flow(&state, flow)?;
|
||||
state.db.replace_flow_outputs(&flow, &schedules, &automations)?;
|
||||
let mut zone_ids = old_zone_ids; zone_ids.extend(schedules.iter().map(|s| s.zone_id.clone()));
|
||||
for zone_id in zone_ids { refresh_zone_override_boundary(&state, &zone_id).await?; }
|
||||
state.log("info", "flow.updated", &format!("Updated Flow {}", flow.name), json!({"flow_id": flow.id, "revision": flow.revision, "schedules": schedules.len(), "automations": automations.len()}));
|
||||
state.broadcast("flow.updated", serde_json::to_value(&flow)?); state.wake_zone_control();
|
||||
Ok(Json(flow))
|
||||
}
|
||||
|
||||
async fn delete_flow(State(state): State<AppState>, Path(id): Path<String>) -> Result<StatusCode, AppError> {
|
||||
let _configuration_guard = state.lock_configuration_operation().await;
|
||||
let _automation_guard = state.lock_automation_operation().await;
|
||||
let _schedule_guard = state.lock_schedule_operation().await;
|
||||
let _cycle_guard = state.lock_zone_control_cycle().await;
|
||||
let zone_ids: std::collections::HashSet<String> = state.db.list_schedules()?.into_iter().filter(|s| s.flow_id.as_deref() == Some(id.as_str())).map(|s| s.zone_id).collect();
|
||||
let existing = state.db.get_flow(&id)?.ok_or_else(|| AppError::NotFound(format!("flow {id}")))?;
|
||||
if !state.db.delete_flow(&id)? { return Err(AppError::NotFound(format!("flow {id}"))); }
|
||||
for zone_id in zone_ids { refresh_zone_override_boundary(&state, &zone_id).await?; }
|
||||
state.log("info", "flow.deleted", &format!("Deleted Flow {}", existing.name), json!({"flow_id": id}));
|
||||
state.broadcast("flow.deleted", json!({"id": id})); state.wake_zone_control();
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
}
|
||||
|
||||
async fn export_flow(State(state): State<AppState>, Path(id): Path<String>) -> Result<Json<Value>, AppError> {
|
||||
let flow = state.db.get_flow(&id)?.ok_or_else(|| AppError::NotFound(format!("flow {id}")))?;
|
||||
Ok(Json(json!({
|
||||
"format": "gree-controller-flow",
|
||||
"version": 1,
|
||||
"exported_at": Utc::now(),
|
||||
"flow": {
|
||||
"name": flow.name,
|
||||
"enabled": flow.enabled,
|
||||
"description": flow.description,
|
||||
"nodes": flow.nodes,
|
||||
"edges": flow.edges
|
||||
}
|
||||
})))
|
||||
}
|
||||
|
||||
async fn import_flow(State(state): State<AppState>, Json(document): Json<Value>) -> Result<(StatusCode, Json<crate::models::Flow>), AppError> {
|
||||
if let Some(format) = document.get("format").and_then(Value::as_str) {
|
||||
if format != "gree-controller-flow" { return Err(AppError::BadRequest("unsupported Flow import format".into())); }
|
||||
}
|
||||
let payload = document.get("flow").cloned().unwrap_or(document);
|
||||
let mut input: FlowInput = serde_json::from_value(payload).map_err(|err| AppError::BadRequest(format!("invalid Flow import: {err}")))?;
|
||||
input.expected_revision = None;
|
||||
validate_flow_graph(&input)?;
|
||||
let _configuration_guard = state.lock_configuration_operation().await;
|
||||
let _automation_guard = state.lock_automation_operation().await;
|
||||
let _schedule_guard = state.lock_schedule_operation().await;
|
||||
let _cycle_guard = state.lock_zone_control_cycle().await;
|
||||
let flow = flow_from_input(Uuid::new_v4().to_string(), input, Utc::now(), 1);
|
||||
let (flow, schedules, automations) = compile_flow(&state, flow)?;
|
||||
state.db.replace_flow_outputs(&flow, &schedules, &automations)?;
|
||||
for zone_id in schedules.iter().map(|s| s.zone_id.as_str()).collect::<std::collections::HashSet<_>>() { refresh_zone_override_boundary(&state, zone_id).await?; }
|
||||
state.log("info", "flow.imported", &format!("Imported Flow {}", flow.name), json!({"flow_id": flow.id, "revision": flow.revision}));
|
||||
state.broadcast("flow.created", serde_json::to_value(&flow)?); state.wake_zone_control();
|
||||
Ok((StatusCode::CREATED, Json(flow)))
|
||||
}
|
||||
|
||||
fn dry_run_block_reason(state: &AppState, action: &crate::models::FlowNode) -> Result<Option<String>, AppError> {
|
||||
match action.kind.as_str() {
|
||||
"zone_thermostat" => {
|
||||
let Some(zone_id) = flow_string(&action.config, "zone_id") else { return Ok(Some("missing_zone".into())); };
|
||||
let Some(zone) = state.db.get_zone(&zone_id)? else { return Ok(Some("missing_zone".into())); };
|
||||
if state.db.get_device(&zone.device_id)?.is_none() { return Ok(Some("missing_device".into())); }
|
||||
if zone.device_manual_override { return Ok(Some("device_manual_override".into())); }
|
||||
if zone.local_thermostat_power.is_some() { return Ok(Some("local_thermostat_override".into())); }
|
||||
if engine::temporary_quick_thermostat_is_active(&zone, Utc::now()) { return Ok(Some("temporary_quick_thermostat".into())); }
|
||||
if !zone.enabled && flow_bool(&action.config, "power") != Some(true) { return Ok(Some("zone_disabled".into())); }
|
||||
Ok(None)
|
||||
}
|
||||
"device_action" => {
|
||||
let Some(device_id) = flow_string(&action.config, "device_id") else { return Ok(Some("missing_device".into())); };
|
||||
let Some(device) = state.db.get_device(&device_id)? else { return Ok(Some("missing_device".into())); };
|
||||
if !device.enabled { return Ok(Some("device_disabled".into())); }
|
||||
let zones = state.db.list_zones()?;
|
||||
if zones.iter().any(|z| z.device_id == device_id && z.device_manual_override) { return Ok(Some("device_manual_override".into())); }
|
||||
if zones.iter().any(|z| z.device_id == device_id && z.local_thermostat_power.is_some()) { return Ok(Some("local_thermostat_override".into())); }
|
||||
if zones.iter().any(|z| z.device_id == device_id && engine::temporary_quick_thermostat_is_active(z, Utc::now())) { return Ok(Some("temporary_quick_thermostat".into())); }
|
||||
if zones.iter().any(|z| z.device_id == device_id && !z.enabled) && flow_bool(&action.config, "power") != Some(true) { return Ok(Some("zone_disabled".into())); }
|
||||
let mut command = DeviceCommand::default();
|
||||
command.power = flow_bool(&action.config, "power");
|
||||
command.mode = flow_string(&action.config, "mode");
|
||||
command.target_temperature = flow_f64(&action.config, "target_temperature");
|
||||
command.fan_speed = flow_u8(&action.config, "fan_speed");
|
||||
command.swing_vertical = flow_bool(&action.config, "swing_vertical");
|
||||
command.swing_horizontal = flow_bool(&action.config, "swing_horizontal");
|
||||
command.quiet = flow_bool(&action.config, "quiet");
|
||||
command.turbo = flow_bool(&action.config, "turbo");
|
||||
command.light = flow_bool(&action.config, "light");
|
||||
command.air = flow_bool(&action.config, "air");
|
||||
command.xfan = flow_bool(&action.config, "xfan");
|
||||
command.health = flow_bool(&action.config, "health");
|
||||
command.sleep = flow_bool(&action.config, "sleep");
|
||||
if zones.iter().any(|z| z.device_id == device_id) && engine::automation_action_conflicts_with_thermostat(&command) {
|
||||
return Ok(Some("thermostat_owner_conflict".into()));
|
||||
}
|
||||
Ok(None)
|
||||
}
|
||||
"group_action" => {
|
||||
let Some(group_id) = flow_string(&action.config, "group_id") else { return Ok(Some("missing_group".into())); };
|
||||
let Some(group) = state.db.get_group(&group_id)? else { return Ok(Some("missing_group".into())); };
|
||||
let climate_change = flow_string(&action.config, "mode").is_some() || flow_string(&action.config, "preset").is_some();
|
||||
let resulting_enabled = flow_bool(&action.config, "power").unwrap_or(group.power_enabled);
|
||||
if climate_change && !resulting_enabled { return Ok(Some("group_control_disabled".into())); }
|
||||
Ok(None)
|
||||
}
|
||||
_ => Ok(Some("unsupported_action".into())),
|
||||
}
|
||||
}
|
||||
|
||||
async fn simulate_flow(State(state): State<AppState>, Json(input): Json<FlowSimulationInput>) -> Result<Json<Value>, AppError> {
|
||||
validate_flow_graph(&input.flow)?;
|
||||
let at = match input.at.as_deref().map(str::trim).filter(|v| !v.is_empty()) {
|
||||
Some(value) => chrono::DateTime::parse_from_rfc3339(value).map_err(|_| AppError::BadRequest("simulation time must be RFC3339".into()))?.with_timezone(&chrono::Local),
|
||||
None => chrono::Local::now(),
|
||||
};
|
||||
let preview_id = input.flow_id.clone().filter(|v| !v.trim().is_empty()).unwrap_or_else(|| format!("dry-run-{}", Uuid::new_v4()));
|
||||
let preview = crate::models::Flow {
|
||||
id: preview_id.clone(), name: input.flow.name.trim().into(), enabled: input.flow.enabled, description: input.flow.description.trim().into(),
|
||||
nodes: input.flow.nodes.clone(), edges: input.flow.edges.clone(), summary: String::new(), compiled_schedule_ids: vec![], compiled_automation_ids: vec![], revision: 0,
|
||||
created_at: Utc::now(), updated_at: Utc::now(),
|
||||
};
|
||||
let (compiled, schedules, automations) = compile_flow(&state, preview)?;
|
||||
let devices = state.db.list_devices()?;
|
||||
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 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!({
|
||||
"node_id": action.id,
|
||||
"kind": action.kind,
|
||||
"matched": matched,
|
||||
"would_execute": matched && blocked_reason.is_none(),
|
||||
"blocked_reason": blocked_reason,
|
||||
"config": action.config,
|
||||
"trace": trace
|
||||
}));
|
||||
}
|
||||
let result = json!({
|
||||
"dry_run": true,
|
||||
"at": at.to_rfc3339(),
|
||||
"flow_id": input.flow_id,
|
||||
"summary": compiled.summary,
|
||||
"compiled": {"schedules": schedules.len(), "automations": automations.len()},
|
||||
"actions": actions,
|
||||
"note": "Dry-run never changes thermostat, device, group, schedule or automation state."
|
||||
});
|
||||
if input.log {
|
||||
state.log("info", "flow.dry_run", &format!("Dry-run Flow {}", input.flow.name.trim()), json!({
|
||||
"flow_id": input.flow_id, "at": at.to_rfc3339(), "actions": result.get("actions").cloned().unwrap_or(Value::Null)
|
||||
}));
|
||||
}
|
||||
Ok(Json(result))
|
||||
}
|
||||
|
||||
async fn flow_logs(State(state): State<AppState>, Path(id): Path<String>, Query(query): Query<FlowLogsQuery>) -> Result<Json<Value>, AppError> {
|
||||
if state.db.get_flow(&id)?.is_none() { return Err(AppError::NotFound(format!("flow {id}"))); }
|
||||
let limit = query.limit.unwrap_or(100).clamp(1, 250) as usize;
|
||||
let events = state.db.list_events(1000)?.into_iter().filter(|event| {
|
||||
event.metadata.get("flow_id").and_then(Value::as_str) == Some(id.as_str())
|
||||
|| event.metadata.get("automation_id").and_then(Value::as_str).map(|value| value.starts_with(&format!("flow:{id}:"))).unwrap_or(false)
|
||||
}).take(limit).collect::<Vec<_>>();
|
||||
Ok(Json(json!({"events": events})))
|
||||
}
|
||||
+1
-1
@@ -256,7 +256,7 @@ async fn apply_schedule_template(State(state): State<AppState>, Path(id): Path<S
|
||||
items.push(Schedule {
|
||||
id: Uuid::new_v4().to_string(), zone_id: id.clone(), name: name.into(), enabled: true,
|
||||
weekdays: days, start_time: start.into(), end_time: end.into(), preset: preset.into(),
|
||||
setpoint: zone.setpoint, created_at: Utc::now(), updated_at: Utc::now(),
|
||||
setpoint: zone.setpoint, created_at: Utc::now(), updated_at: Utc::now(), flow_id: None, flow_node_id: None,
|
||||
});
|
||||
};
|
||||
let all = vec![1,2,3,4,5,6,7];
|
||||
|
||||
@@ -25,7 +25,7 @@ impl ScheduleInput {
|
||||
fn into_schedule(self, id: String, created_at: chrono::DateTime<Utc>) -> Schedule {
|
||||
Schedule { id, zone_id: self.zone_id, name: self.name.trim().into(), enabled: self.enabled,
|
||||
weekdays: self.weekdays, start_time: self.start_time, end_time: self.end_time,
|
||||
preset: self.preset, setpoint: self.setpoint, created_at, updated_at: Utc::now() }
|
||||
preset: self.preset, setpoint: self.setpoint, created_at, updated_at: Utc::now(), flow_id: None, flow_node_id: None }
|
||||
}
|
||||
}
|
||||
fn validate_schedule_set(items: &[Schedule]) -> Result<(), AppError> {
|
||||
@@ -119,6 +119,7 @@ async fn update_schedule(State(state): State<AppState>, Path(id): Path<String>,
|
||||
let _cycle_guard = state.lock_zone_control_cycle().await;
|
||||
input.validate()?;
|
||||
let existing = state.db.get_schedule(&id)?.ok_or_else(|| AppError::NotFound(format!("schedule {id}")))?;
|
||||
if existing.flow_id.is_some() { return Err(AppError::BadRequest("this schedule is generated by Flow; edit it in the Flow editor".into())); }
|
||||
if state.db.get_zone(&input.zone_id)?.is_none() { return Err(AppError::BadRequest("schedule zone does not exist".into())); }
|
||||
let old_zone_id = existing.zone_id.clone();
|
||||
let item = input.into_schedule(id.clone(), existing.created_at);
|
||||
@@ -135,6 +136,7 @@ async fn delete_schedule(State(state): State<AppState>, Path(id): Path<String>)
|
||||
let _schedule_guard = state.lock_schedule_operation().await;
|
||||
let _cycle_guard = state.lock_zone_control_cycle().await;
|
||||
let existing = state.db.get_schedule(&id)?.ok_or_else(|| AppError::NotFound(format!("schedule {id}")))?;
|
||||
if existing.flow_id.is_some() { return Err(AppError::BadRequest("this schedule is generated by Flow; delete it from the Flow editor".into())); }
|
||||
if !state.db.delete_schedule(&id)? { return Err(AppError::NotFound(format!("schedule {id}"))); }
|
||||
refresh_zone_override_boundary(&state, &existing.zone_id).await?;
|
||||
state.broadcast("schedule.deleted", json!({"id": id}));
|
||||
|
||||
+36
-9
@@ -154,7 +154,7 @@ async fn export_settings(State(state): State<AppState>) -> Result<Json<Configura
|
||||
}
|
||||
|
||||
fn validate_configuration_export(export: &ConfigurationExport) -> Result<(), AppError> {
|
||||
if export.format_version != 1 { return Err(AppError::BadRequest("unsupported configuration export version".into())); }
|
||||
if !matches!(export.format_version, 1 | 2) { return Err(AppError::BadRequest("unsupported configuration export version".into())); }
|
||||
influxdb::validate(&export.settings.influxdb).map_err(|err| AppError::BadRequest(err.to_string()))?;
|
||||
if !matches!(export.settings.house_mode.as_str(), "cool" | "heat" | "off") {
|
||||
return Err(AppError::BadRequest("import contains an invalid house mode".into()));
|
||||
@@ -164,9 +164,10 @@ fn validate_configuration_export(export: &ConfigurationExport) -> Result<(), App
|
||||
let zones: std::collections::HashSet<&str> = export.zones.iter().map(|item| item.id.as_str()).collect();
|
||||
let schedules: std::collections::HashSet<&str> = export.schedules.iter().map(|item| item.id.as_str()).collect();
|
||||
let automations: std::collections::HashSet<&str> = export.automations.iter().map(|item| item.id.as_str()).collect();
|
||||
let flows: std::collections::HashSet<&str> = export.flows.iter().map(|item| item.id.as_str()).collect();
|
||||
if devices.len() != export.devices.len() || zones.len() != export.zones.len()
|
||||
|| schedules.len() != export.schedules.len() || automations.len() != export.automations.len()
|
||||
|| devices.contains("") || zones.contains("") || schedules.contains("") || automations.contains("")
|
||||
|| schedules.len() != export.schedules.len() || automations.len() != export.automations.len() || flows.len() != export.flows.len()
|
||||
|| devices.contains("") || zones.contains("") || schedules.contains("") || automations.contains("") || flows.contains("")
|
||||
{
|
||||
return Err(AppError::BadRequest("import contains duplicate or empty resource IDs".into()));
|
||||
}
|
||||
@@ -194,6 +195,9 @@ fn validate_configuration_export(export: &ConfigurationExport) -> Result<(), App
|
||||
return Err(AppError::BadRequest("import contains a schedule referencing a missing zone".into()));
|
||||
}
|
||||
for item in &export.schedules {
|
||||
if item.flow_id.as_deref().is_some_and(|flow_id| !flows.contains(flow_id)) {
|
||||
return Err(AppError::BadRequest("import contains a Flow-generated schedule referencing a missing Flow".into()));
|
||||
}
|
||||
if item.weekdays.is_empty() || item.weekdays.iter().any(|day| !(1..=7).contains(day)) {
|
||||
return Err(AppError::BadRequest("import contains invalid schedule weekdays".into()));
|
||||
}
|
||||
@@ -234,10 +238,27 @@ fn validate_configuration_export(export: &ConfigurationExport) -> Result<(), App
|
||||
let at = item.at_time.as_deref().ok_or_else(|| AppError::BadRequest("import contains a time automation without at_time".into()))?;
|
||||
NaiveTime::parse_from_str(at, "%H:%M").map_err(|_| AppError::BadRequest("import contains an invalid automation time".into()))?;
|
||||
}
|
||||
"flow" => {
|
||||
if item.flow_id.as_deref().filter(|id| flows.contains(*id)).is_none() || item.flow_conditions.is_empty() {
|
||||
return Err(AppError::BadRequest("import contains an invalid Flow-generated automation".into()));
|
||||
}
|
||||
}
|
||||
_ => return Err(AppError::BadRequest("import contains an unsupported automation trigger".into())),
|
||||
}
|
||||
|
||||
if let Some(group_id) = item.action_group_id.as_deref().filter(|value| !value.is_empty()) {
|
||||
if let Some(zone_id) = item.action_zone_id.as_deref().filter(|value| !value.is_empty()) {
|
||||
if !zones.contains(zone_id) { return Err(AppError::BadRequest("import contains a Flow automation referencing a missing zone".into())); }
|
||||
if let Some(preset) = item.action_zone_preset.as_deref() {
|
||||
if !matches!(preset, "auto" | "custom" | "comfort" | "sleep" | "away") {
|
||||
return Err(AppError::BadRequest("import contains an invalid Flow thermostat preset".into()));
|
||||
}
|
||||
}
|
||||
if item.action_zone_preset.as_deref() == Some("custom")
|
||||
&& item.action.target_temperature.is_some_and(|value| !(8.0..=30.0).contains(&value))
|
||||
{
|
||||
return Err(AppError::BadRequest("import contains an invalid Flow thermostat target".into()));
|
||||
}
|
||||
} else if let Some(group_id) = item.action_group_id.as_deref().filter(|value| !value.is_empty()) {
|
||||
if !groups.contains(group_id) {
|
||||
return Err(AppError::BadRequest("import contains an automation referencing a missing group".into()));
|
||||
}
|
||||
@@ -246,17 +267,23 @@ fn validate_configuration_export(export: &ConfigurationExport) -> Result<(), App
|
||||
return Err(AppError::BadRequest("import contains an invalid group automation mode".into()));
|
||||
}
|
||||
}
|
||||
let flow_custom_group = item.flow_id.is_some() && item.action_preset.as_deref() == Some("custom");
|
||||
if let Some(preset) = item.action_preset.as_deref() {
|
||||
if !matches!(preset, "auto" | "comfort" | "sleep" | "away") {
|
||||
if !matches!(preset, "auto" | "comfort" | "sleep" | "away") && !(flow_custom_group && preset == "custom") {
|
||||
return Err(AppError::BadRequest("import contains an invalid group automation preset".into()));
|
||||
}
|
||||
}
|
||||
if item.action.target_temperature.is_some() || item.action.fan_speed.is_some()
|
||||
|| item.action.swing_vertical.is_some() || item.action.swing_horizontal.is_some()
|
||||
if flow_custom_group {
|
||||
let Some(target) = item.action.target_temperature else { return Err(AppError::BadRequest("import contains a Flow custom group preset without a target".into())); };
|
||||
if !(8.0..=30.0).contains(&target) { return Err(AppError::BadRequest("import contains an invalid Flow group target".into())); }
|
||||
} else if item.action.target_temperature.is_some() {
|
||||
return Err(AppError::BadRequest("import contains unsupported target temperature in a group automation".into()));
|
||||
}
|
||||
if item.action.fan_speed.is_some() || item.action.swing_vertical.is_some() || item.action.swing_horizontal.is_some()
|
||||
|| item.action.quiet.is_some() || item.action.turbo.is_some() || item.action.light.is_some()
|
||||
|| item.action.air.is_some() || item.action.xfan.is_some() || item.action.health.is_some() || item.action.sleep.is_some()
|
||||
{
|
||||
return Err(AppError::BadRequest("import contains unsupported fields in a group automation".into()));
|
||||
return Err(AppError::BadRequest("import contains unsupported device fields in a group automation".into()));
|
||||
}
|
||||
if item.action.power.is_none() && item.action.mode.is_none() && item.action_preset.as_deref().filter(|v| !v.is_empty()).is_none() {
|
||||
return Err(AppError::BadRequest("import contains an empty group automation action".into()));
|
||||
@@ -361,7 +388,7 @@ async fn import_settings(State(state): State<AppState>, Json(mut export): Json<C
|
||||
|
||||
// Configuration replacement is the broadest mutation in the application. Serialize it
|
||||
// against every structural editor and every live owner that can write zone/device state.
|
||||
// Global lock order: configuration -> automation -> house -> schedule -> zones -> devices.
|
||||
// Global lock order: configuration -> automation -> house -> schedule -> thermostat-cycle -> zones -> devices.
|
||||
let _configuration_guard = state.lock_configuration_operation().await;
|
||||
let _automation_guard = state.lock_automation_operation().await;
|
||||
let _house_guard = state.lock_house_operation().await;
|
||||
|
||||
@@ -26,6 +26,7 @@ async fn build_bootstrap(state: &AppState) -> Result<Value, AppError> {
|
||||
"groups": state.db.list_groups()?,
|
||||
"schedules": state.db.list_schedules()?,
|
||||
"automations": state.db.list_automations()?,
|
||||
"flows": state.db.list_flows()?,
|
||||
"access_tokens": state.db.list_api_tokens()?,
|
||||
"settings": public_settings(&settings),
|
||||
"outdoor_temperature": *state.outdoor_temperature.read().await,
|
||||
|
||||
@@ -5,7 +5,7 @@ use rusqlite::{params, Connection, OptionalExtension};
|
||||
use serde::{de::DeserializeOwned, Serialize};
|
||||
use serde_json::Value;
|
||||
use crate::{
|
||||
models::{ApiTokenInfo, Automation, ClimateGroup, ConfigurationExport, Device, EventLog, HaReading, Reading, RuntimeSettings, Schedule, Zone, ZoneReading},
|
||||
models::{ApiTokenInfo, Automation, ClimateGroup, ConfigurationExport, Device, EventLog, Flow, HaReading, Reading, RuntimeSettings, Schedule, Zone, ZoneReading},
|
||||
queries,
|
||||
};
|
||||
|
||||
@@ -19,6 +19,7 @@ pub struct Db {
|
||||
include!("db/core_devices.rs");
|
||||
include!("db/climate.rs");
|
||||
include!("db/schedules_automations.rs");
|
||||
include!("db/flows.rs");
|
||||
include!("db/device_history.rs");
|
||||
include!("db/zone_history.rs");
|
||||
include!("db/ha_history.rs");
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
impl Db {
|
||||
pub fn export_configuration(&self, settings: RuntimeSettings) -> Result<ConfigurationExport> {
|
||||
Ok(ConfigurationExport {
|
||||
format_version: 1,
|
||||
format_version: 2,
|
||||
exported_at: Utc::now(),
|
||||
settings,
|
||||
devices: self.list_devices()?,
|
||||
@@ -9,6 +9,7 @@ impl Db {
|
||||
groups: self.list_groups()?,
|
||||
schedules: self.list_schedules()?,
|
||||
automations: self.list_automations()?,
|
||||
flows: self.list_flows()?,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -36,6 +37,10 @@ impl Db {
|
||||
let payload = Self::to_json(item)?;
|
||||
tx.execute(queries::UPSERT_AUTOMATION, params![item.id, payload, item.updated_at.to_rfc3339()])?;
|
||||
}
|
||||
for flow in &export.flows {
|
||||
let payload = Self::to_json(flow)?;
|
||||
tx.execute(queries::UPSERT_FLOW, params![flow.id, payload, flow.updated_at.to_rfc3339()])?;
|
||||
}
|
||||
let settings_json = Self::to_json(&export.settings)?;
|
||||
tx.execute(queries::UPSERT_RUNTIME_SETTINGS, params![settings_json, Utc::now().to_rfc3339()])?;
|
||||
tx.commit()?;
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
impl Db {
|
||||
pub fn list_flows(&self) -> Result<Vec<Flow>> {
|
||||
self.list_payloads(queries::LIST_FLOWS)
|
||||
}
|
||||
|
||||
pub fn get_flow(&self, id: &str) -> Result<Option<Flow>> {
|
||||
self.get_payload(queries::GET_FLOW, id)
|
||||
}
|
||||
|
||||
pub fn replace_flow_outputs(&self, flow: &Flow, schedules: &[Schedule], automations: &[Automation]) -> Result<()> {
|
||||
let mut conn = self.lock()?;
|
||||
let tx = conn.transaction()?;
|
||||
tx.execute(queries::DELETE_SCHEDULES_BY_FLOW_ID, [flow.id.as_str()])?;
|
||||
tx.execute(queries::DELETE_AUTOMATIONS_BY_FLOW_ID, [flow.id.as_str()])?;
|
||||
for schedule in schedules {
|
||||
let payload = Self::to_json(schedule)?;
|
||||
tx.execute(queries::UPSERT_SCHEDULE, params![schedule.id, schedule.zone_id, payload, schedule.updated_at.to_rfc3339()])?;
|
||||
}
|
||||
for item in automations {
|
||||
let payload = Self::to_json(item)?;
|
||||
tx.execute(queries::UPSERT_AUTOMATION, params![item.id, payload, item.updated_at.to_rfc3339()])?;
|
||||
}
|
||||
let payload = Self::to_json(flow)?;
|
||||
tx.execute(queries::UPSERT_FLOW, params![flow.id, payload, flow.updated_at.to_rfc3339()])?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn delete_flow(&self, id: &str) -> Result<bool> {
|
||||
let mut conn = self.lock()?;
|
||||
let tx = conn.transaction()?;
|
||||
tx.execute(queries::DELETE_SCHEDULES_BY_FLOW_ID, [id])?;
|
||||
tx.execute(queries::DELETE_AUTOMATIONS_BY_FLOW_ID, [id])?;
|
||||
let changed = tx.execute(queries::DELETE_FLOW, [id])? > 0;
|
||||
tx.commit()?;
|
||||
Ok(changed)
|
||||
}
|
||||
}
|
||||
+411
-22
@@ -1,14 +1,15 @@
|
||||
|
||||
pub fn automation_action_conflicts_with_thermostat(command: &DeviceCommand) -> bool {
|
||||
// Power/mode/target are translated by apply_automatic_device_action into durable zone
|
||||
// state, so they do not fight the thermostat. Fan/quiet/sleep are thermostat outputs with
|
||||
// no independent zone override model; accepting them as one-shot direct automation would
|
||||
// let the next thermostat cycle immediately overwrite them.
|
||||
command.fan_speed.is_some() || command.quiet.is_some() || command.sleep.is_some()
|
||||
// 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_enabled_thermostat_zone(device_id: &str, zones: &[Zone]) -> bool {
|
||||
zones.iter().any(|zone| zone.device_id == device_id && zone.enabled)
|
||||
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<()> {
|
||||
@@ -27,6 +28,13 @@ 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 {
|
||||
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,
|
||||
};
|
||||
if !should_fire { continue; }
|
||||
@@ -44,41 +52,49 @@ async fn run_automations(state: &AppState) -> Result<()> {
|
||||
let zones = state.db.list_zones()?;
|
||||
let groups = state.db.list_groups()?;
|
||||
|
||||
if item.action_group_id.is_none()
|
||||
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
|
||||
"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() && device_blocked_by_manual_override(&item.action_device_id, &zones) {
|
||||
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
|
||||
"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() && device_blocked_by_local_thermostat(&item.action_device_id, &zones) {
|
||||
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
|
||||
"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()
|
||||
&& device_has_enabled_thermostat_zone(&item.action_device_id, &zones)
|
||||
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
|
||||
"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 let Some(group_id) = item.action_group_id.as_deref() {
|
||||
let target_devices: Vec<String> = 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()))
|
||||
@@ -90,6 +106,8 @@ async fn run_automations(state: &AppState) -> Result<()> {
|
||||
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,
|
||||
@@ -97,20 +115,22 @@ async fn run_automations(state: &AppState) -> Result<()> {
|
||||
continue;
|
||||
}
|
||||
|
||||
let result: Result<bool, AppError> = if let Some(group_id) = item.action_group_id.as_deref() {
|
||||
let result: Result<bool, AppError> = 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: None,
|
||||
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
|
||||
"automation_id": item.id, "device_id": item.action_device_id, "flow_id": item.flow_id, "flow_node_id": item.flow_node_id
|
||||
}));
|
||||
Ok(false)
|
||||
}
|
||||
@@ -124,12 +144,18 @@ async fn run_automations(state: &AppState) -> Result<()> {
|
||||
item.updated_at = Utc::now();
|
||||
state.db.save_automation(&item)?;
|
||||
state.log("info", "automation.fired", &format!("Automation {} fired", item.name), json!({
|
||||
"automation_id": item.id, "group_id": item.action_group_id, "device_id": item.action_device_id
|
||||
"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
|
||||
@@ -137,7 +163,7 @@ async fn run_automations(state: &AppState) -> Result<()> {
|
||||
item.last_fired_at = Some(Utc::now());
|
||||
item.updated_at = Utc::now();
|
||||
state.db.save_automation(&item)?;
|
||||
state.log("error", "automation.error", &err.to_string(), json!({"automation_id": item.id}));
|
||||
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}));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -156,6 +182,11 @@ 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.
|
||||
@@ -182,3 +213,361 @@ fn time_automation_due(item: &Automation, now: DateTime<Local>) -> bool {
|
||||
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,
|
||||
})
|
||||
}
|
||||
|
||||
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 c = &condition.config;
|
||||
let override_value = overrides.get(&condition.id).cloned();
|
||||
let (matched, actual) = match condition.kind.as_str() {
|
||||
"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))
|
||||
}
|
||||
"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>,
|
||||
) -> 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 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()))
|
||||
}
|
||||
_ 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);
|
||||
final_id = Some(condition.id.clone());
|
||||
trace.push(json!({
|
||||
"node_id": condition.id,
|
||||
"kind": condition.kind,
|
||||
"matched": matched,
|
||||
"actual": actual,
|
||||
"expected": condition.config.get("value"),
|
||||
"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], conditions: &[crate::models::FlowCondition]) -> Result<bool, AppError> {
|
||||
let overrides = HashMap::new();
|
||||
evaluate_flow_conditions_trace(state, devices, conditions, Local::now(), &overrides).await.map(|(matched, _)| matched)
|
||||
}
|
||||
|
||||
fn flow_condition_kind_runtime(kind: &str) -> bool {
|
||||
matches!(kind,
|
||||
"weekday" | "time_range" | "date_range" | "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"
|
||||
)
|
||||
}
|
||||
|
||||
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((target.clamp(8.0,30.0)*2.0).round()/2.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, true);
|
||||
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)?); state.wake_zone_control();
|
||||
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.
|
||||
let _device_guard = state.lock_device_operation(&device_id).await;
|
||||
send_command_locked_forced(state, &device_id, DeviceCommand { power: Some(false), ..Default::default() }).await?;
|
||||
}
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ pub async fn build_control_plan(state: &AppState) -> Result<ControlPlan, AppErro
|
||||
let devices = state.db.list_devices()?;
|
||||
let zones = state.db.list_zones()?;
|
||||
let groups = state.db.list_groups()?;
|
||||
let zone_names: HashMap<String, String> = zones.iter().map(|zone| (zone.id.clone(), zone.name.clone())).collect();
|
||||
let house_preset = zones.first().and_then(|first| {
|
||||
let first_preset = first.manual_preset.as_deref().unwrap_or("auto");
|
||||
zones.iter().all(|zone| zone.manual_preset.as_deref().unwrap_or("auto") == first_preset)
|
||||
@@ -111,10 +112,13 @@ pub async fn build_control_plan(state: &AppState) -> Result<ControlPlan, AppErro
|
||||
}
|
||||
let mut rules = Vec::new();
|
||||
for item in state.db.list_automations()? {
|
||||
let action_zone_name = item.action_zone_id.as_deref()
|
||||
.and_then(|id| zone_names.get(id))
|
||||
.cloned();
|
||||
let action_group_name = item.action_group_id.as_deref()
|
||||
.and_then(|id| groups.iter().find(|group| group.id == id))
|
||||
.map(|group| group.name.clone());
|
||||
let action_name = action_group_name.clone().unwrap_or_else(|| {
|
||||
let action_name = action_zone_name.or_else(|| action_group_name.clone()).unwrap_or_else(|| {
|
||||
devices.iter().find(|device| device.id == item.action_device_id)
|
||||
.map(|device| device.name.clone())
|
||||
.unwrap_or_else(|| item.action_device_id.clone())
|
||||
|
||||
+14
-3
@@ -17,7 +17,7 @@ mod tests {
|
||||
let item = Schedule {
|
||||
id: "1".into(), zone_id: "z".into(), name: "night".into(), enabled: true,
|
||||
weekdays: vec![1], start_time: "22:00".into(), end_time: "06:00".into(), preset: "custom".into(), setpoint: 20.0,
|
||||
created_at: Utc::now(), updated_at: Utc::now(),
|
||||
created_at: Utc::now(), updated_at: Utc::now(), flow_id: None, flow_node_id: None,
|
||||
};
|
||||
assert!(schedule_active(&item, now));
|
||||
}
|
||||
@@ -26,7 +26,7 @@ mod tests {
|
||||
Schedule {
|
||||
id: id.into(), zone_id: "z".into(), name: id.into(), enabled: true,
|
||||
weekdays, start_time: start.into(), end_time: end.into(), preset: "comfort".into(), setpoint: 21.0,
|
||||
created_at: Utc::now(), updated_at: Utc::now(),
|
||||
created_at: Utc::now(), updated_at: Utc::now(), flow_id: None, flow_node_id: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -94,7 +94,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, created_at: Utc::now(), updated_at: Utc::now(),
|
||||
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(),
|
||||
};
|
||||
assert!(time_automation_due(&item, now.clone()));
|
||||
item.last_fired_at = Some((now.clone() - chrono::Duration::seconds(35)).with_timezone(&Utc));
|
||||
@@ -173,6 +173,9 @@ mod tests {
|
||||
assert!(!automation_action_conflicts_with_thermostat(&DeviceCommand { target_temperature: Some(22.0), ..Default::default() }));
|
||||
assert!(automation_action_conflicts_with_thermostat(&DeviceCommand { fan_speed: Some(3), ..Default::default() }));
|
||||
assert!(automation_action_conflicts_with_thermostat(&DeviceCommand { quiet: Some(true), ..Default::default() }));
|
||||
assert!(automation_action_conflicts_with_thermostat(&DeviceCommand { mode: Some("fan".into()), ..Default::default() }));
|
||||
assert!(automation_action_conflicts_with_thermostat(&DeviceCommand { mode: Some("dry".into()), ..Default::default() }));
|
||||
assert!(!automation_action_conflicts_with_thermostat(&DeviceCommand { mode: Some("heat".into()), ..Default::default() }));
|
||||
assert!(!automation_action_conflicts_with_thermostat(&DeviceCommand { light: Some(false), turbo: Some(true), ..Default::default() }));
|
||||
}
|
||||
|
||||
@@ -195,6 +198,14 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn active_temporary_thermostat_blocks_direct_automation() {
|
||||
let now = Utc::now();
|
||||
let mut zone = test_zone("device");
|
||||
zone.temporary_quick_thermostat = Some(temporary_session(now));
|
||||
assert!(device_blocked_by_temporary_thermostat(&zone.device_id, std::slice::from_ref(&zone)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn local_thermostat_resume_clears_only_local_quick_control_state() {
|
||||
let mut zone = test_zone("device");
|
||||
|
||||
@@ -49,6 +49,7 @@ async fn send_automatic_device_command_if_owned(
|
||||
if device_blocked_by_disabled_zone(device_id, &zones)
|
||||
|| device_blocked_by_manual_override(device_id, &zones)
|
||||
|| device_blocked_by_local_thermostat(device_id, &zones)
|
||||
|| device_blocked_by_temporary_thermostat(device_id, &zones)
|
||||
{
|
||||
return Ok(None);
|
||||
}
|
||||
@@ -63,6 +64,9 @@ async fn apply_automatic_device_action(
|
||||
// Direct automation target/setpoint is translated into zone state and may use the next
|
||||
// schedule boundary. Serialize that derivation with schedule edits before locking the zone.
|
||||
let _schedule_guard = state.lock_schedule_operation().await;
|
||||
// Device automations can alter durable thermostat state. Serialize that transition with
|
||||
// the thermostat cycle so it cannot act on a snapshot taken before this automation.
|
||||
let _cycle_guard = state.lock_zone_control_cycle().await;
|
||||
let zones = state.db.list_zones()?;
|
||||
let Some(zone_id) = zones.iter().find(|zone| zone.device_id == device_id).map(|zone| zone.id.clone()) else {
|
||||
return send_automatic_device_command_if_owned(state, device_id, command).await;
|
||||
@@ -71,7 +75,7 @@ async fn apply_automatic_device_action(
|
||||
let _zone_guard = state.lock_zone_operation(&zone_id).await;
|
||||
let _device_guard = state.lock_device_operation(device_id).await;
|
||||
let mut zone = state.db.get_zone(&zone_id)?.ok_or_else(|| AppError::NotFound(format!("zone {zone_id}")))?;
|
||||
if zone.device_manual_override || zone.local_thermostat_power.is_some() {
|
||||
if zone.device_manual_override || zone.local_thermostat_power.is_some() || temporary_quick_thermostat_is_active(&zone, Utc::now()) {
|
||||
return Ok(None);
|
||||
}
|
||||
// A power-on automation is an explicit domain transition and may re-enable a zone that
|
||||
|
||||
@@ -674,6 +674,9 @@ async fn control_zones(state: &AppState) -> Result<()> {
|
||||
"quiet": updated_device.quiet,
|
||||
"sleep": updated_device.sleep,
|
||||
"night_mode": night_active,
|
||||
"schedule_id": active_schedule.map(|item| item.id.as_str()),
|
||||
"flow_id": active_schedule.and_then(|item| item.flow_id.as_deref()),
|
||||
"flow_node_id": active_schedule.and_then(|item| item.flow_node_id.as_deref()),
|
||||
}));
|
||||
}
|
||||
Ok(None) => {
|
||||
|
||||
@@ -107,3 +107,40 @@ mod tests {
|
||||
assert_eq!(resolve_entity_id(&settings, None).as_deref(), Some("sensor.salon_temperature"));
|
||||
}
|
||||
}
|
||||
|
||||
/// Read a complete Home Assistant entity document for Flow conditions. Keeping this helper
|
||||
/// centralized means state and attribute blocks share the same URL validation, TLS and auth path.
|
||||
pub async fn read_entity(
|
||||
default_client: &reqwest::Client,
|
||||
settings: &HomeAssistantSettings,
|
||||
entity_override: Option<&str>,
|
||||
) -> 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") }
|
||||
let entity = resolve_entity_id(settings, entity_override)
|
||||
.ok_or_else(|| anyhow!("Home Assistant entity_id is not configured"))?;
|
||||
let mut base = Url::parse(settings.url.trim()).context("invalid Home Assistant URL")?;
|
||||
if !matches!(base.scheme(), "http" | "https") { bail!("Home Assistant URL must use http or https") }
|
||||
base = base.join(&format!("api/states/{}", entity.trim_start_matches('/'))).context("cannot build Home Assistant API URL")?;
|
||||
let client = request_client(default_client, settings)?;
|
||||
let response = client.get(base).bearer_auth(settings.token.trim()).header("Accept", "application/json")
|
||||
.send().await.context("Home Assistant 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>())
|
||||
}
|
||||
response.json().await.context("invalid Home Assistant JSON")
|
||||
}
|
||||
|
||||
/// Read the raw Home Assistant state for Flow conditions. Unlike `read_temperature`, this
|
||||
/// intentionally keeps the state as text so binary_sensor, switch, input_boolean and custom
|
||||
/// entities can participate in visual automations.
|
||||
pub async fn read_state(
|
||||
default_client: &reqwest::Client,
|
||||
settings: &HomeAssistantSettings,
|
||||
entity_override: Option<&str>,
|
||||
) -> Result<String> {
|
||||
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"))
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ include!("models/defaults.rs");
|
||||
include!("models/device.rs");
|
||||
include!("models/temporary_thermostat.rs");
|
||||
include!("models/zone.rs");
|
||||
include!("models/flow.rs");
|
||||
include!("models/automation.rs");
|
||||
include!("models/history.rs");
|
||||
include!("models/integrations.rs");
|
||||
|
||||
@@ -17,6 +17,12 @@ pub struct Schedule {
|
||||
pub setpoint: f64,
|
||||
pub created_at: DateTime<Utc>,
|
||||
pub updated_at: DateTime<Utc>,
|
||||
/// Present for records generated by the visual Flow editor. Generated records are read-only
|
||||
/// in the legacy schedule/automation editors; the Flow remains the source of truth.
|
||||
#[serde(default)]
|
||||
pub flow_id: Option<String>,
|
||||
#[serde(default)]
|
||||
pub flow_node_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
@@ -25,7 +31,7 @@ pub struct Automation {
|
||||
pub name: String,
|
||||
#[serde(default = "default_true")]
|
||||
pub enabled: bool,
|
||||
/// temperature_above, temperature_below, time
|
||||
/// temperature_above, temperature_below, time, flow
|
||||
pub trigger_kind: String,
|
||||
#[serde(default)]
|
||||
pub trigger_device_id: Option<String>,
|
||||
@@ -48,6 +54,18 @@ pub struct Automation {
|
||||
pub cooldown_seconds: u64,
|
||||
#[serde(default)]
|
||||
pub last_fired_at: Option<DateTime<Utc>>,
|
||||
/// Optional Flow-owned thermostat zone target.
|
||||
#[serde(default)]
|
||||
pub action_zone_id: Option<String>,
|
||||
#[serde(default)]
|
||||
pub action_zone_preset: Option<String>,
|
||||
/// Topologically ordered condition/logic program compiled from the visual Flow graph.
|
||||
#[serde(default)]
|
||||
pub flow_conditions: Vec<FlowCondition>,
|
||||
#[serde(default)]
|
||||
pub flow_id: Option<String>,
|
||||
#[serde(default)]
|
||||
pub flow_node_id: Option<String>,
|
||||
pub created_at: DateTime<Utc>,
|
||||
pub updated_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
@@ -9,6 +9,8 @@ pub struct ConfigurationExport {
|
||||
pub groups: Vec<ClimateGroup>,
|
||||
pub schedules: Vec<Schedule>,
|
||||
pub automations: Vec<Automation>,
|
||||
#[serde(default)]
|
||||
pub flows: Vec<Flow>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct FlowNode {
|
||||
pub id: String,
|
||||
pub kind: String,
|
||||
#[serde(default)]
|
||||
pub x: f64,
|
||||
#[serde(default)]
|
||||
pub y: f64,
|
||||
#[serde(default)]
|
||||
pub config: Value,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct FlowEdge {
|
||||
pub id: String,
|
||||
pub from: String,
|
||||
pub to: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct FlowCondition {
|
||||
/// Stable block ID used by the compiled Flow program.
|
||||
#[serde(default)]
|
||||
pub id: String,
|
||||
pub kind: String,
|
||||
#[serde(default)]
|
||||
pub config: Value,
|
||||
/// IDs of predecessor blocks. Condition blocks implicitly AND their own predicate with
|
||||
/// predecessor results; logic blocks combine predecessors according to their kind.
|
||||
#[serde(default)]
|
||||
pub inputs: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Flow {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
#[serde(default = "default_true")]
|
||||
pub enabled: bool,
|
||||
#[serde(default)]
|
||||
pub description: String,
|
||||
#[serde(default)]
|
||||
pub nodes: Vec<FlowNode>,
|
||||
#[serde(default)]
|
||||
pub edges: Vec<FlowEdge>,
|
||||
#[serde(default)]
|
||||
pub summary: String,
|
||||
#[serde(default)]
|
||||
pub compiled_schedule_ids: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub compiled_automation_ids: Vec<String>,
|
||||
/// Optimistic concurrency token for visual-editor saves. Legacy flows deserialize as 0.
|
||||
#[serde(default)]
|
||||
pub revision: u64,
|
||||
pub created_at: DateTime<Utc>,
|
||||
pub updated_at: DateTime<Utc>,
|
||||
}
|
||||
@@ -68,3 +68,16 @@ pub const LIST_AUTOMATIONS: &str =
|
||||
pub const GET_AUTOMATION: &str = "SELECT payload FROM automations WHERE id=?1";
|
||||
pub const DELETE_AUTOMATION: &str = "DELETE FROM automations WHERE id=?1";
|
||||
|
||||
|
||||
|
||||
pub const UPSERT_FLOW: &str = r#"
|
||||
INSERT INTO flows(id,payload,updated_at) VALUES(?1,?2,?3)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
payload=excluded.payload,
|
||||
updated_at=excluded.updated_at
|
||||
"#;
|
||||
pub const LIST_FLOWS: &str = "SELECT payload FROM flows ORDER BY json_extract(payload, '$.name') COLLATE NOCASE";
|
||||
pub const GET_FLOW: &str = "SELECT payload FROM flows WHERE id=?1";
|
||||
pub const DELETE_FLOW: &str = "DELETE FROM flows WHERE id=?1";
|
||||
pub const DELETE_SCHEDULES_BY_FLOW_ID: &str = "DELETE FROM schedules WHERE json_extract(payload, '$.flow_id')=?1";
|
||||
pub const DELETE_AUTOMATIONS_BY_FLOW_ID: &str = "DELETE FROM automations WHERE json_extract(payload, '$.flow_id')=?1";
|
||||
|
||||
@@ -37,6 +37,7 @@ DELETE FROM ha_readings WHERE id IN (
|
||||
pub const CLEAR_CONFIGURATION: &str = r#"
|
||||
DELETE FROM schedules;
|
||||
DELETE FROM automations;
|
||||
DELETE FROM flows;
|
||||
DELETE FROM climate_groups;
|
||||
DELETE FROM zones;
|
||||
DELETE FROM devices;
|
||||
|
||||
@@ -50,6 +50,12 @@ CREATE TABLE IF NOT EXISTS automations (
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS flows (
|
||||
id TEXT PRIMARY KEY,
|
||||
payload TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS readings (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
device_id TEXT NOT NULL,
|
||||
@@ -127,5 +133,7 @@ INSERT OR IGNORE INTO schema_migrations(version, applied_at)
|
||||
VALUES (4, strftime('%Y-%m-%dT%H:%M:%fZ','now'));
|
||||
INSERT OR IGNORE INTO schema_migrations(version, applied_at)
|
||||
VALUES (5, strftime('%Y-%m-%dT%H:%M:%fZ','now'));
|
||||
INSERT OR IGNORE INTO schema_migrations(version, applied_at)
|
||||
VALUES (6, strftime('%Y-%m-%dT%H:%M:%fZ','now'));
|
||||
"#;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user