gru/gree-controller
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0148daf41021965a3bcbf3d092192453ae0dd0f9
gree-controller/src/api/flows.rs
T
Mateusz Gruszczyński be3708f14d v0.15.8
2026-09-18 23:57:28 +02:00

2223 lines
84 KiB
Rust
Raw Blame History

#[derive(Debug, Deserialize)]
struct FlowInput {
name: String,
#[serde(default = "yes")]
enabled: bool,
#[serde(default)]
draft: 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_referenced_shared_input_ids(
nodes: &[crate::models::FlowNode],
) -> std::collections::HashSet<String> {
nodes
.iter()
.filter(|node| node.kind == "shared_input")
.filter_map(|node| flow_string(&node.config, "input_id"))
.collect()
}
fn same_shared_input_source(
left: &crate::models::FlowSharedInput,
right: &crate::models::FlowSharedInput,
) -> bool {
left.kind == right.kind && left.config == right.config
}
fn prepare_imported_shared_inputs(
input: &mut FlowInput,
imported: Vec<crate::models::FlowSharedInput>,
current: &[crate::models::FlowSharedInput],
) -> Result<(Vec<crate::models::FlowSharedInput>, bool), AppError> {
let mut imported_ids = std::collections::HashSet::new();
let mut normalized_imported = Vec::with_capacity(imported.len());
for mut item in imported {
item.id = item.id.trim().chars().take(120).collect();
item.name = item.name.trim().chars().take(100).collect();
item.kind = item.kind.trim().to_string();
if item.id.is_empty() || item.name.is_empty() {
return Err(AppError::BadRequest(
"imported shared Flow input requires id and name".into(),
));
}
if !imported_ids.insert(item.id.clone()) {
return Err(AppError::BadRequest(
"imported shared Flow input IDs must be unique".into(),
));
}
normalized_imported.push(item);
}
let referenced_import_ids = flow_referenced_shared_input_ids(&input.nodes);
let mut next = current.to_vec();
let mut remap = std::collections::HashMap::<String, String>::new();
let mut changed = false;
for item in normalized_imported
.into_iter()
.filter(|item| referenced_import_ids.contains(&item.id))
{
if let Some(existing) = next.iter().find(|existing| existing.id == item.id) {
if same_shared_input_source(existing, &item) {
remap.insert(item.id.clone(), existing.id.clone());
continue;
}
}
if let Some(existing) = next
.iter()
.find(|existing| same_shared_input_source(existing, &item))
{
remap.insert(item.id.clone(), existing.id.clone());
continue;
}
let original_id = item.id.clone();
let mut to_add = item;
if next.iter().any(|existing| existing.id == to_add.id) {
to_add.id = format!("shared-{}", Uuid::new_v4());
}
remap.insert(original_id, to_add.id.clone());
next.push(to_add);
changed = true;
}
for node in &mut input.nodes {
if node.kind != "shared_input" {
continue;
}
let Some(input_id) = flow_string(&node.config, "input_id") else {
continue;
};
if let Some(mapped) = remap.get(&input_id) {
if mapped != &input_id {
let config = node.config.as_object_mut().ok_or_else(|| {
AppError::BadRequest("shared Flow input block config must be an object".into())
})?;
config.insert("input_id".into(), Value::String(mapped.clone()));
}
}
}
for input_id in flow_referenced_shared_input_ids(&input.nodes) {
if !next.iter().any(|item| item.id == input_id) {
return Err(AppError::BadRequest(format!(
"shared Flow input block references missing input {input_id}; export the Flow again with shared input definitions or create the input first"
)));
}
}
Ok((next, changed))
}
fn flow_louver_position(config: &Value, key: &str) -> Option<u8> {
match config.get(key) {
Some(Value::Bool(value)) => Some(u8::from(*value)),
Some(value) => value
.as_u64()
.and_then(|raw| u8::try_from(raw).ok()),
None => None,
}
}
fn flow_device_feature_command(config: &Value) -> Result<DeviceCommand, AppError> {
let feature = flow_string(config, "feature")
.ok_or_else(|| AppError::BadRequest("device feature action needs a feature".into()))?;
let mut command = DeviceCommand::default();
match feature.as_str() {
"power" => {
command.power = Some(flow_bool(config, "value").ok_or_else(|| {
AppError::BadRequest("device feature power needs a boolean value".into())
})?);
}
"mode" => {
command.mode = Some(flow_string(config, "value").ok_or_else(|| {
AppError::BadRequest("device feature mode needs a value".into())
})?);
}
"target_temperature" => {
command.target_temperature = Some(flow_f64(config, "value").ok_or_else(|| {
AppError::BadRequest("device feature target temperature needs a numeric value".into())
})?);
}
"fan_speed" => {
command.fan_speed = Some(flow_u8(config, "value").ok_or_else(|| {
AppError::BadRequest("device feature fan speed needs an integer value".into())
})?);
}
"swing_vertical" => {
command.swing_vertical = Some(flow_louver_position(config, "value").ok_or_else(|| {
AppError::BadRequest("device feature vertical louver needs a position".into())
})?);
}
"swing_horizontal" => {
command.swing_horizontal = Some(flow_louver_position(config, "value").ok_or_else(|| {
AppError::BadRequest("device feature horizontal louver needs a position".into())
})?);
}
"quiet" | "turbo" | "light" | "air" | "xfan" | "health" | "sleep" => {
let value = flow_bool(config, "value").ok_or_else(|| {
AppError::BadRequest("device feature action needs an on/off value".into())
})?;
match feature.as_str() {
"quiet" => command.quiet = Some(value),
"turbo" => command.turbo = Some(value),
"light" => command.light = Some(value),
"air" => command.air = Some(value),
"xfan" => command.xfan = Some(value),
"health" => command.health = Some(value),
"sleep" => command.sleep = Some(value),
_ => unreachable!(),
}
}
_ => {
return Err(AppError::BadRequest(format!(
"unsupported device feature action: {feature}"
)))
}
}
engine::validate_command(&command)?;
Ok(command)
}
fn generated_flow_name(flow_id: &str, action_node_id: &str) -> String {
let digest = Sha256::digest(format!("{flow_id}:{action_node_id}").as_bytes());
format!("flow-{}", URL_SAFE_NO_PAD.encode(&digest[..12]))
}
fn flow_condition_kind(kind: &str) -> bool {
matches!(
kind,
"weekday"
| "time_range"
| "date_range"
| "cron_trigger"
| "stable_for"
| "delay"
| "state_duration"
| "on_change"
| "rate_limit"
| "rolling_stat"
| "oscillates"
| "outdoor_temperature"
| "device_temperature"
| "zone_temperature"
| "ha_state"
| "ha_numeric"
| "ha_attribute"
| "ha_available"
| "house_mode"
| "device_state"
| "zone_state"
| "group_state"
| "night_mode"
| "constant"
| "shared_input"
)
}
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"
| "device_feature_action"
| "group_action"
| "ha_service_action"
)
}
fn shared_input_comparison_kind(kind: &str) -> bool {
matches!(
kind,
"outdoor_temperature"
| "device_temperature"
| "zone_temperature"
| "ha_state"
| "ha_numeric"
| "ha_attribute"
| "house_mode"
| "device_state"
| "zone_state"
| "group_state"
)
}
enum SharedInputResourceReference {
Device(String),
Zone(String),
Group(String),
}
fn shared_input_resource_reference(
kind: &str,
config: &Value,
) -> Result<Option<SharedInputResourceReference>, AppError> {
let reference = match kind {
"outdoor_temperature" | "house_mode" | "night_mode" => None,
"device_temperature" => {
let id = flow_string(config, "device_id").ok_or_else(|| {
AppError::BadRequest("shared device temperature input needs device".into())
})?;
Some(SharedInputResourceReference::Device(id))
}
"zone_temperature" => {
let id = flow_string(config, "zone_id").ok_or_else(|| {
AppError::BadRequest("shared zone temperature input needs zone".into())
})?;
Some(SharedInputResourceReference::Zone(id))
}
"ha_state" | "ha_numeric" | "ha_available" => {
if flow_string(config, "entity_id").is_none() {
return Err(AppError::BadRequest(
"shared Home Assistant input needs entity_id".into(),
));
}
None
}
"ha_attribute" => {
if flow_string(config, "entity_id").is_none()
|| flow_string(config, "attribute").is_none()
{
return Err(AppError::BadRequest(
"shared Home Assistant attribute input needs entity_id and attribute".into(),
));
}
None
}
"device_state" => {
let id = flow_string(config, "device_id").ok_or_else(|| {
AppError::BadRequest("shared device state input needs device".into())
})?;
let field = flow_string(config, "field").ok_or_else(|| {
AppError::BadRequest("shared device state input 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 shared device state field".into(),
));
}
Some(SharedInputResourceReference::Device(id))
}
"zone_state" => {
let id = flow_string(config, "zone_id")
.ok_or_else(|| AppError::BadRequest("shared zone state input needs zone".into()))?;
let field = flow_string(config, "field").ok_or_else(|| {
AppError::BadRequest("shared zone state input 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 shared zone state field".into(),
));
}
Some(SharedInputResourceReference::Zone(id))
}
"group_state" => {
let id = flow_string(config, "group_id").ok_or_else(|| {
AppError::BadRequest("shared group state input needs group".into())
})?;
if flow_string(config, "field").as_deref() != Some("power_enabled") {
return Err(AppError::BadRequest(
"unsupported shared group state field".into(),
));
}
Some(SharedInputResourceReference::Group(id))
}
"constant" => {
if config.get("value").and_then(Value::as_bool).is_none() {
return Err(AppError::BadRequest(
"shared constant input needs a boolean value".into(),
));
}
None
}
_ => {
return Err(AppError::BadRequest(format!(
"unsupported shared Flow input kind: {kind}"
)))
}
};
Ok(reference)
}
fn validate_shared_input_source(
kind: &str,
config: &Value,
state: &AppState,
) -> Result<(), AppError> {
let Some(reference) = shared_input_resource_reference(kind, config)? else {
return Ok(());
};
let exists = match reference {
SharedInputResourceReference::Device(id) => state.db.get_device(&id)?.is_some(),
SharedInputResourceReference::Zone(id) => state.db.get_zone(&id)?.is_some(),
SharedInputResourceReference::Group(id) => state.db.get_group(&id)?.is_some(),
};
if !exists {
return Err(AppError::BadRequest(
"shared Flow input references a missing resource".into(),
));
}
Ok(())
}
fn validate_flow_draft_graph(input: &FlowInput) -> Result<(), AppError> {
if input.name.trim().is_empty() {
return Err(AppError::BadRequest("flow name is required".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
)));
}
}
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 !ids.contains(edge.from.as_str()) || !ids.contains(edge.to.as_str()) {
return Err(AppError::BadRequest(
"flow contains a connection to a missing block".into(),
));
}
}
Ok(())
}
fn validate_flow_graph(input: &FlowInput) -> Result<(), AppError> {
validate_flow_draft_graph(input)?;
if input.nodes.is_empty() {
return Err(AppError::BadRequest("flow needs at least one block".into()));
}
let ids = input
.nodes
.iter()
.map(|node| node.id.clone())
.collect::<std::collections::HashSet<_>>();
if !input.nodes.iter().any(|node| flow_action_kind(&node.kind)) {
return Err(AppError::BadRequest(
"flow needs at least one action block".into(),
));
}
for edge in &input.edges {
if edge.from == edge.to {
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(),
));
}
}
for node in input.nodes.iter().filter(|node| node.kind == "rate_limit") {
let targets = input
.edges
.iter()
.filter(|edge| edge.from == node.id)
.filter_map(|edge| input.nodes.iter().find(|target| target.id == edge.to))
.collect::<Vec<_>>();
if targets.is_empty() || targets.iter().any(|target| !flow_action_kind(&target.kind)) {
return Err(AppError::BadRequest(
"rate-limit block must be placed directly before an action".into(),
));
}
}
for node in input.nodes.iter().filter(|node| {
node.kind == "on_change" && flow_string(&node.config, "mode").as_deref() == Some("value")
}) {
let sources = input
.edges
.iter()
.filter(|edge| edge.to == node.id)
.filter_map(|edge| input.nodes.iter().find(|source| source.id == edge.from))
.collect::<Vec<_>>();
if sources.len() != 1
|| sources.iter().any(|source| {
flow_logic_kind(&source.kind)
|| matches!(
source.kind.as_str(),
"stable_for"
| "delay"
| "state_duration"
| "on_change"
| "rate_limit"
| "rolling_stat"
| "oscillates"
)
})
{
return Err(AppError::BadRequest(
"on-change value mode needs one direct source/condition input".into(),
));
}
}
// Reject cycles for executable Flows. Drafts intentionally allow unfinished wiring and are never compiled.
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,
shared_inputs_override: Option<&[crate::models::FlowSharedInput]>,
) -> 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(),
));
}
}
"cron_trigger" => {
let expr = flow_string(&condition.config, "expression")
.ok_or_else(|| AppError::BadRequest("cron block needs an expression".into()))?;
if !engine::cron_expression_valid(&expr) {
return Err(AppError::BadRequest(
"invalid cron expression; expected 5 fields with *, */N, ranges or lists"
.into(),
));
}
}
"stable_for" => {
let seconds = condition
.config
.get("seconds")
.and_then(Value::as_u64)
.unwrap_or(0);
if seconds == 0 || seconds > 604800 {
return Err(AppError::BadRequest(
"stable-for duration must be between 1 second and 7 days".into(),
));
}
if condition.inputs.len() != 1 {
return Err(AppError::BadRequest(
"stable-for block needs exactly one input".into(),
));
}
}
"delay" => {
let seconds = condition
.config
.get("seconds")
.and_then(Value::as_u64)
.unwrap_or(0);
if seconds == 0 || seconds > 604800 {
return Err(AppError::BadRequest(
"delay must be between 1 second and 7 days".into(),
));
}
if condition.inputs.len() != 1 {
return Err(AppError::BadRequest(
"delay block needs exactly one input".into(),
));
}
}
"state_duration" => {
let min_seconds = condition
.config
.get("min_seconds")
.and_then(Value::as_u64)
.unwrap_or(0);
let max_seconds = condition.config.get("max_seconds").and_then(Value::as_u64);
if min_seconds > 604800 || max_seconds.is_some_and(|value| value > 604800) {
return Err(AppError::BadRequest(
"state duration must be between 0 seconds and 7 days".into(),
));
}
if max_seconds.is_some_and(|value| value < min_seconds) {
return Err(AppError::BadRequest(
"state duration maximum must be greater than or equal to minimum".into(),
));
}
if min_seconds == 0 && max_seconds.is_none() {
return Err(AppError::BadRequest(
"state duration needs a minimum or maximum duration".into(),
));
}
if condition.inputs.len() != 1 {
return Err(AppError::BadRequest(
"state duration block needs exactly one input".into(),
));
}
}
"on_change" => {
let mode = flow_string(&condition.config, "mode").unwrap_or_else(|| "result".into());
if !matches!(mode.as_str(), "result" | "value") {
return Err(AppError::BadRequest(
"on-change mode must be result or value".into(),
));
}
if condition.inputs.len() != 1 {
return Err(AppError::BadRequest(
"on-change block needs exactly one input".into(),
));
}
}
"rate_limit" => {
let max_count = condition
.config
.get("max_count")
.and_then(Value::as_u64)
.unwrap_or(0);
let period_seconds = condition
.config
.get("period_seconds")
.and_then(Value::as_u64)
.unwrap_or(0);
if !(1..=1000).contains(&max_count) {
return Err(AppError::BadRequest(
"rate limit max_count must be between 1 and 1000".into(),
));
}
if !(1..=2678400).contains(&period_seconds) {
return Err(AppError::BadRequest(
"rate limit period must be between 1 second and 31 days".into(),
));
}
if condition.inputs.len() != 1 {
return Err(AppError::BadRequest(
"rate-limit block needs exactly one input".into(),
));
}
}
"rolling_stat" => {
let source = flow_string(&condition.config, "source").unwrap_or_default();
if !matches!(
source.as_str(),
"outdoor_temperature" | "device_temperature" | "zone_temperature" | "ha_numeric"
) {
return Err(AppError::BadRequest(
"rolling statistic has an unsupported source".into(),
));
}
if source == "device_temperature" {
let id = flow_string(&condition.config, "device_id").unwrap_or_default();
if state.db.get_device(&id)?.is_none() {
return Err(AppError::BadRequest(
"rolling statistic references a missing device".into(),
));
}
}
if source == "zone_temperature" {
let id = flow_string(&condition.config, "zone_id").unwrap_or_default();
if state.db.get_zone(&id)?.is_none() {
return Err(AppError::BadRequest(
"rolling statistic references a missing zone".into(),
));
}
}
if source == "ha_numeric" && flow_string(&condition.config, "entity_id").is_none() {
return Err(AppError::BadRequest(
"rolling HA statistic needs entity_id".into(),
));
}
let window = condition
.config
.get("window_seconds")
.and_then(Value::as_u64)
.unwrap_or(0);
if window < 10 || window > 604800 {
return Err(AppError::BadRequest(
"rolling statistic window must be between 10 seconds and 7 days".into(),
));
}
if !matches!(
flow_string(&condition.config, "statistic").as_deref(),
Some("mean") | Some("median")
) {
return Err(AppError::BadRequest(
"rolling statistic must be mean or median".into(),
));
}
validate_flow_comparison(&condition.config)?;
}
"oscillates" => {
let source = flow_string(&condition.config, "source").unwrap_or_default();
if !matches!(
source.as_str(),
"outdoor_temperature" | "device_temperature" | "zone_temperature" | "ha_numeric"
) {
return Err(AppError::BadRequest(
"oscillation block has an unsupported source".into(),
));
}
if source == "device_temperature" {
let id = flow_string(&condition.config, "device_id").unwrap_or_default();
if state.db.get_device(&id)?.is_none() {
return Err(AppError::BadRequest(
"oscillation block references a missing device".into(),
));
}
}
if source == "zone_temperature" {
let id = flow_string(&condition.config, "zone_id").unwrap_or_default();
if state.db.get_zone(&id)?.is_none() {
return Err(AppError::BadRequest(
"oscillation block references a missing zone".into(),
));
}
}
if source == "ha_numeric" && flow_string(&condition.config, "entity_id").is_none() {
return Err(AppError::BadRequest(
"oscillation HA source needs entity_id".into(),
));
}
let window = condition
.config
.get("window_seconds")
.and_then(Value::as_u64)
.unwrap_or(0);
if window < 10 || window > 604800 {
return Err(AppError::BadRequest(
"oscillation window must be between 10 seconds and 7 days".into(),
));
}
let min_span = flow_f64(&condition.config, "min_span").unwrap_or(0.0);
if !min_span.is_finite() || min_span <= 0.0 {
return Err(AppError::BadRequest(
"oscillation minimum span must be greater than zero".into(),
));
}
let min_changes = condition
.config
.get("min_direction_changes")
.and_then(Value::as_u64)
.unwrap_or(0);
if min_changes == 0 || min_changes > 1000 {
return Err(AppError::BadRequest(
"oscillation direction changes must be between 1 and 1000".into(),
));
}
}
"outdoor_temperature" => {
validate_flow_comparison(&condition.config)?;
}
"device_temperature" => {
validate_flow_comparison(&condition.config)?;
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(),
));
}
}
"shared_input" => {
let input_id = flow_string(&condition.config, "input_id").ok_or_else(|| {
AppError::BadRequest("shared Flow input block needs input_id".into())
})?;
let shared = if let Some(inputs) = shared_inputs_override {
inputs.iter().find(|item| item.id == input_id).cloned()
} else {
state
.db
.load_runtime_settings()?
.and_then(|settings| {
settings
.home_assistant
.flow_inputs
.into_iter()
.find(|item| item.id == input_id)
})
}
.ok_or_else(|| {
AppError::BadRequest(format!(
"shared Flow input block references missing input {input_id}"
))
})?;
validate_shared_input_source(&shared.kind, &shared.config, state)?;
if shared_input_comparison_kind(&shared.kind) {
let operator = flow_string(&condition.config, "operator").ok_or_else(|| {
AppError::BadRequest(
"shared Flow value needs an operator in the Flow block".into(),
)
})?;
let mut config = shared.config.clone();
let map = config.as_object_mut().ok_or_else(|| {
AppError::BadRequest("shared Flow input config must be an object".into())
})?;
map.insert("operator".into(), Value::String(operator));
map.insert(
"value".into(),
condition
.config
.get("value")
.cloned()
.unwrap_or(Value::Null),
);
let resolved = crate::models::FlowCondition {
id: condition.id.clone(),
kind: shared.kind.clone(),
config,
inputs: Vec::new(),
};
validate_condition(&resolved, state, shared_inputs_override)?;
} else if flow_string(&condition.config, "operator").is_some()
|| condition.config.get("value").is_some()
{
return Err(AppError::BadRequest(
"this shared Flow input is already boolean and does not accept a comparison"
.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,
shared_inputs_override: Option<&[crate::models::FlowSharedInput]>,
) -> 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, shared_inputs_override)?;
}
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()
&& flow_louver_position(&action_node.config, "swing_vertical").is_none()
&& flow_louver_position(&action_node.config, "swing_horizontal").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: generated_flow_name(&flow.id, &action_node.id),
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 flow_runtime = previous_automations
.get(&id)
.map(|item| {
let previous_conditions = item
.flow_conditions
.iter()
.map(|condition| (condition.id.as_str(), condition))
.collect::<std::collections::HashMap<_, _>>();
let current_conditions = conditions
.iter()
.map(|condition| (condition.id.as_str(), condition))
.collect::<std::collections::HashMap<_, _>>();
let mut runtime = item.flow_runtime.clone();
runtime.retain(|node_id, _| {
let Some(previous) = previous_conditions.get(node_id.as_str()) else {
return false;
};
let Some(current) = current_conditions.get(node_id.as_str()) else {
return false;
};
matches!(
current.kind.as_str(),
"stable_for"
| "delay"
| "state_duration"
| "on_change"
| "rate_limit"
| "rolling_stat"
| "oscillates"
) && previous.kind == current.kind
&& previous.config == current.config
&& previous.inputs == current.inputs
});
runtime
})
.unwrap_or_default();
let mut item = Automation {
id,
name: generated_flow_name(&flow.id, &action_node.id),
enabled: flow.enabled,
trigger_kind: "flow".into(),
trigger_device_id: None,
threshold: None,
at_time: None,
action_device_id: String::new(),
action_group_id: None,
action_preset: None,
action: DeviceCommand::default(),
cooldown_seconds: 60,
last_fired_at,
action_zone_id: None,
action_zone_preset: None,
action_ha_domain: None,
action_ha_service: None,
action_ha_entity_id: None,
action_ha_data: Value::Null,
flow_conditions: conditions,
flow_id: Some(flow.id.clone()),
flow_node_id: Some(action_node.id.clone()),
flow_runtime,
created_at,
updated_at: now.clone(),
};
if let Some(cooldown) = action_node
.config
.get("cooldown_seconds")
.and_then(Value::as_u64)
{
item.cooldown_seconds = cooldown.max(30);
}
match action_node.kind.as_str() {
"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);
}
item.action.swing_vertical =
flow_louver_position(&action_node.config, "swing_vertical");
item.action.swing_horizontal =
flow_louver_position(&action_node.config, "swing_horizontal");
engine::validate_command(&item.action)?;
}
"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_louver_position(&action_node.config, "swing_vertical");
item.action.swing_horizontal =
flow_louver_position(&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()));
}
}
"device_feature_action" => {
let id = flow_string(&action_node.config, "device_id").ok_or_else(|| {
AppError::BadRequest("device feature action needs a device".into())
})?;
if !devices.iter().any(|d| d.id == id) {
return Err(AppError::BadRequest(
"device feature action references a missing device".into(),
));
}
item.action_device_id = id;
item.action = flow_device_feature_command(&action_node.config)?;
}
"ha_service_action" => {
let domain = flow_string(&action_node.config, "domain").ok_or_else(|| {
AppError::BadRequest("Home Assistant action needs a domain".into())
})?;
let service = flow_string(&action_node.config, "service").ok_or_else(|| {
AppError::BadRequest("Home Assistant action needs a service".into())
})?;
item.action_ha_domain = Some(domain);
item.action_ha_service = Some(service);
item.action_ha_entity_id = flow_string(&action_node.config, "entity_id");
item.action_ha_data = action_node
.config
.get("data")
.cloned()
.unwrap_or_else(|| json!({}));
if !item.action_ha_data.is_object() {
return Err(AppError::BadRequest(
"Home Assistant service data must be a JSON object".into(),
));
}
}
"group_action" => {
let id = flow_string(&action_node.config, "group_id")
.ok_or_else(|| AppError::BadRequest("group action needs a group".into()))?;
if !groups.iter().any(|g| g.id == id) {
return Err(AppError::BadRequest(
"group action references a missing group".into(),
));
}
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 {
let draft = input.draft;
crate::models::Flow {
id,
name: input.name.trim().into(),
enabled: if draft { false } else { input.enabled },
draft,
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(),
}
}
fn prepare_draft_flow(mut flow: crate::models::Flow) -> crate::models::Flow {
flow.enabled = false;
flow.draft = true;
flow.compiled_schedule_ids.clear();
flow.compiled_automation_ids.clear();
flow.summary = format!("Draft · {} blocks", flow.nodes.len());
flow
}
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> {
if input.draft {
validate_flow_draft_graph(&input)?;
} else {
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) = if flow.draft {
(prepare_draft_flow(flow), vec![], vec![])
} else {
compile_flow(&state, flow, None)?
};
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> {
if input.draft {
validate_flow_draft_graph(&input)?;
} else {
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) = if flow.draft {
(prepare_draft_flow(flow), vec![], vec![])
} else {
compile_flow(&state, flow, None)?
};
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}")))?;
let referenced = flow_referenced_shared_input_ids(&flow.nodes);
let shared_inputs: Vec<_> = state
.settings
.read()
.await
.home_assistant
.flow_inputs
.iter()
.filter(|item| referenced.contains(&item.id))
.cloned()
.collect();
Ok(Json(json!({
"format": "gree-controller-flow",
"version": 1,
"exported_at": Utc::now(),
"shared_inputs": shared_inputs,
"flow": {
"name": flow.name,
"enabled": flow.enabled,
"draft": flow.draft,
"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(),
));
}
}
if let Some(version) = document.get("version").and_then(Value::as_u64) {
if !matches!(version, 1 | 2) {
return Err(AppError::BadRequest(format!(
"unsupported Flow import version {version}"
)));
}
}
let imported_shared_inputs: Vec<crate::models::FlowSharedInput> = document
.get("shared_inputs")
.cloned()
.map(serde_json::from_value)
.transpose()
.map_err(|err| AppError::BadRequest(format!("invalid shared Flow inputs: {err}")))?
.unwrap_or_default();
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;
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 mut next_settings = state.settings.read().await.clone();
let (flow_inputs, shared_inputs_changed) = prepare_imported_shared_inputs(
&mut input,
imported_shared_inputs,
&next_settings.home_assistant.flow_inputs,
)?;
next_settings.home_assistant.flow_inputs = flow_inputs;
validate_flow_shared_inputs(&mut next_settings.home_assistant, &state)?;
if input.draft {
validate_flow_draft_graph(&input)?;
} else {
validate_flow_graph(&input)?;
}
let flow = flow_from_input(Uuid::new_v4().to_string(), input, Utc::now(), 1);
let (flow, schedules, automations) = if flow.draft {
(prepare_draft_flow(flow), vec![], vec![])
} else {
compile_flow(
&state,
flow,
Some(&next_settings.home_assistant.flow_inputs),
)?
};
if shared_inputs_changed {
state.db.replace_flow_outputs_with_runtime_settings(
&flow,
&schedules,
&automations,
&next_settings,
)?;
*state.settings.write().await = next_settings.clone();
let payload = home_assistant_settings(&next_settings);
state.broadcast(
"settings.home_assistant.updated",
serde_json::to_value(&payload)?,
);
} else {
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,
"shared_inputs_imported": shared_inputs_changed
}),
);
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" | "device_feature_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()));
}
let command = if action.kind == "device_feature_action" {
flow_device_feature_command(&action.config)?
} else {
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_louver_position(&action.config, "swing_vertical");
command.swing_horizontal = flow_louver_position(&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");
command
};
if zones.iter().any(|z| z.device_id == device_id && !z.enabled)
&& command.power != Some(true)
{
return Ok(Some("zone_disabled".into()));
}
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)
}
"ha_service_action" => 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,
draft: false,
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, None)?;
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 automation_id = format!("flow:{preview_id}:automation:{}", action.id);
let mut runtime = state
.db
.get_automation(&automation_id)?
.map(|item| item.flow_runtime)
.unwrap_or_default();
let (matched, trace) = engine::evaluate_flow_conditions_trace(
&state,
&devices,
&program,
at.clone(),
&input.overrides,
Some(&mut runtime),
)
.await?;
let blocked_reason = if matched && !input.flow.enabled {
Some("flow_disabled".into())
} else if matched {
dry_run_block_reason(&state, action)?
} else {
None
};
actions.push(json!({
"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})))
}
#[cfg(test)]
mod flow_draft_tests {
use super::*;
fn unfinished_input() -> FlowInput {
FlowInput {
name: "Unfinished".into(),
enabled: true,
draft: true,
description: String::new(),
nodes: vec![crate::models::FlowNode {
id: "condition".into(),
kind: "constant".into(),
x: 0.0,
y: 0.0,
config: json!({"value": true}),
}],
edges: vec![],
expected_revision: None,
}
}
#[test]
fn draft_graph_accepts_flow_without_action() {
let input = unfinished_input();
assert!(validate_flow_draft_graph(&input).is_ok());
assert!(validate_flow_graph(&input).is_err());
}
#[test]
fn draft_graph_still_rejects_missing_edge_endpoints() {
let mut input = unfinished_input();
input.edges.push(crate::models::FlowEdge {
id: "edge".into(),
from: "condition".into(),
to: "missing".into(),
});
assert!(validate_flow_draft_graph(&input).is_err());
}
#[test]
fn draft_flow_is_forced_disabled_and_has_no_compiled_outputs() {
let flow = flow_from_input("draft-id".into(), unfinished_input(), chrono::Utc::now(), 1);
let flow = prepare_draft_flow(flow);
assert!(flow.draft);
assert!(!flow.enabled);
assert!(flow.compiled_schedule_ids.is_empty());
assert!(flow.compiled_automation_ids.is_empty());
}
#[test]
fn device_feature_action_builds_only_selected_field() {
let command = flow_device_feature_command(&json!({"feature": "light", "value": true}))
.expect("valid feature action");
assert_eq!(command.light, Some(true));
assert!(command.power.is_none());
assert!(command.mode.is_none());
assert!(command.swing_vertical.is_none());
}
#[test]
fn device_feature_action_validates_louver_range() {
let result = flow_device_feature_command(&json!({"feature": "swing_vertical", "value": 99}));
assert!(result.is_err());
}
}
Reference in New Issue View Git Blame Copy Permalink