v0.12.0-preety_code

src/influxdb/codec.rs

@@ -20,30 +20,100 @@ fn parse_csv_line(line: &str) -> Vec<String> {
     out
 }
 
-fn flux_query(settings: &InfluxDbSettings, measurement: &str, extra_filters: &str, group_tags: &[&str], start: DateTime<Utc>, stop: DateTime<Utc>, bucket_seconds: i64) -> String {
-    let tags = group_tags.iter().map(|tag| format!("\"{tag}\"")).collect::<Vec<_>>().join(",");
+fn flux_query(
+    settings: &InfluxDbSettings,
+    measurement: &str,
+    extra_filters: &str,
+    group_tags: &[&str],
+    start: DateTime<Utc>,
+    stop: DateTime<Utc>,
+    bucket_seconds: i64,
+) -> String {
+    let tags = group_tags
+        .iter()
+        .map(|tag| format!("\"{tag}\""))
+        .collect::<Vec<_>>()
+        .join(",");
     format!(
         "from(bucket: {}) |> range(start: time(v: {}), stop: time(v: {})) |> filter(fn: (r) => r._measurement == {}){} |> aggregateWindow(every: {}s, fn: mean, createEmpty: false) |> group(columns: [{}]) |> pivot(rowKey:[\"_time\"], columnKey:[\"_field\"], valueColumn:\"_value\") |> sort(columns:[\"_time\"])",
         flux_string(&settings.bucket), flux_string(&start.to_rfc3339()), flux_string(&stop.to_rfc3339()), flux_string(measurement), extra_filters, bucket_seconds.max(1), tags
     )
 }
 
-fn line_protocol(measurement: &str, tags: &[(&str, &str)], fields: Vec<String>, timestamp: DateTime<Utc>) -> Result<String> {
-    if fields.is_empty() { bail!("InfluxDB measurement has no fields"); }
-    let tags = tags.iter().filter(|(_, value)| !value.is_empty()).map(|(key,value)| format!(",{}={}", escape_tag(key), escape_tag(value))).collect::<String>();
-    let nanos = timestamp.timestamp_nanos_opt().ok_or_else(|| anyhow!("timestamp outside nanosecond range"))?;
-    Ok(format!("{}{} {} {}", escape_measurement(measurement), tags, fields.join(","), nanos))
+fn line_protocol(
+    measurement: &str,
+    tags: &[(&str, &str)],
+    fields: Vec<String>,
+    timestamp: DateTime<Utc>,
+) -> Result<String> {
+    if fields.is_empty() {
+        bail!("InfluxDB measurement has no fields");
+    }
+    let tags = tags
+        .iter()
+        .filter(|(_, value)| !value.is_empty())
+        .map(|(key, value)| format!(",{}={}", escape_tag(key), escape_tag(value)))
+        .collect::<String>();
+    let nanos = timestamp
+        .timestamp_nanos_opt()
+        .ok_or_else(|| anyhow!("timestamp outside nanosecond range"))?;
+    Ok(format!(
+        "{}{} {} {}",
+        escape_measurement(measurement),
+        tags,
+        fields.join(","),
+        nanos
+    ))
 }
 
-fn push_float(fields: &mut Vec<String>, key: &str, value: Option<f64>) { if let Some(value) = value.filter(|v| v.is_finite()) { fields.push(format!("{}={value}", escape_field_key(key))); } }
-fn push_int(fields: &mut Vec<String>, key: &str, value: i64) { fields.push(format!("{}={value}i", escape_field_key(key))); }
-fn escape_measurement(value: &str) -> String { value.replace('\\', "\\\\").replace(',', "\\,").replace(' ', "\\ ") }
-fn escape_tag(value: &str) -> String { value.replace('\\', "\\\\").replace(',', "\\,").replace('=', "\\=").replace(' ', "\\ ") }
-fn escape_field_key(value: &str) -> String { escape_tag(value) }
-fn influxql_string(value: &str) -> String { value.replace('\\', "\\\\").replace('\'', "\\'") }
-fn flux_string(value: &str) -> String { format!("\"{}\"", value.replace('\\', "\\\\").replace('"', "\\\"")) }
-fn truncate(value: &str, max: usize) -> String { value.chars().take(max).collect() }
-fn row_f64(row: &HashMap<String,String>, key: &str) -> Option<f64> { row.get(key)?.parse().ok() }
-fn parse_flux_time(row: &HashMap<String,String>) -> Option<DateTime<Utc>> { DateTime::parse_from_rfc3339(row.get("_time")?).ok().map(|v| v.with_timezone(&Utc)) }
-fn row_num(row: &HashMap<String,Value>, key: &str) -> Option<f64> { row.get(key)?.as_f64().or_else(|| row.get(key)?.as_i64().map(|v|v as f64)) }
-fn row_time(row: &HashMap<String,Value>) -> Option<DateTime<Utc>> { DateTime::parse_from_rfc3339(row.get("time")?.as_str()?).ok().map(|v|v.with_timezone(&Utc)) }
+fn push_float(fields: &mut Vec<String>, key: &str, value: Option<f64>) {
+    if let Some(value) = value.filter(|v| v.is_finite()) {
+        fields.push(format!("{}={value}", escape_field_key(key)));
+    }
+}
+fn push_int(fields: &mut Vec<String>, key: &str, value: i64) {
+    fields.push(format!("{}={value}i", escape_field_key(key)));
+}
+fn escape_measurement(value: &str) -> String {
+    value
+        .replace('\\', "\\\\")
+        .replace(',', "\\,")
+        .replace(' ', "\\ ")
+}
+fn escape_tag(value: &str) -> String {
+    value
+        .replace('\\', "\\\\")
+        .replace(',', "\\,")
+        .replace('=', "\\=")
+        .replace(' ', "\\ ")
+}
+fn escape_field_key(value: &str) -> String {
+    escape_tag(value)
+}
+fn influxql_string(value: &str) -> String {
+    value.replace('\\', "\\\\").replace('\'', "\\'")
+}
+fn flux_string(value: &str) -> String {
+    format!("\"{}\"", value.replace('\\', "\\\\").replace('"', "\\\""))
+}
+fn truncate(value: &str, max: usize) -> String {
+    value.chars().take(max).collect()
+}
+fn row_f64(row: &HashMap<String, String>, key: &str) -> Option<f64> {
+    row.get(key)?.parse().ok()
+}
+fn parse_flux_time(row: &HashMap<String, String>) -> Option<DateTime<Utc>> {
+    DateTime::parse_from_rfc3339(row.get("_time")?)
+        .ok()
+        .map(|v| v.with_timezone(&Utc))
+}
+fn row_num(row: &HashMap<String, Value>, key: &str) -> Option<f64> {
+    row.get(key)?
+        .as_f64()
+        .or_else(|| row.get(key)?.as_i64().map(|v| v as f64))
+}
+fn row_time(row: &HashMap<String, Value>) -> Option<DateTime<Utc>> {
+    DateTime::parse_from_rfc3339(row.get("time")?.as_str()?)
+        .ok()
+        .map(|v| v.with_timezone(&Utc))
+}

src/influxdb/query.rs

@@ -8,15 +8,43 @@ pub async fn query_devices(
     limit: u32,
 ) -> Result<Vec<Reading>> {
     if settings.version == "1" {
-        query_devices_v1(client, settings, device_id, start, stop, bucket_seconds, limit).await
+        query_devices_v1(
+            client,
+            settings,
+            device_id,
+            start,
+            stop,
+            bucket_seconds,
+            limit,
+        )
+        .await
     } else {
-        let tags = if let Some(value) = device_id { format!(" |> filter(fn: (r) => r.device_id == {})", flux_string(value)) } else { String::new() };
-        let query = flux_query(settings, DEVICE_MEASUREMENT, &tags, &["device_id"], start, stop, bucket_seconds);
+        let tags = if let Some(value) = device_id {
+            format!(
+                " |> filter(fn: (r) => r.device_id == {})",
+                flux_string(value)
+            )
+        } else {
+            String::new()
+        };
+        let query = flux_query(
+            settings,
+            DEVICE_MEASUREMENT,
+            &tags,
+            &["device_id"],
+            start,
+            stop,
+            bucket_seconds,
+        );
         let rows = query_v2(client, settings, &query).await?;
         let mut out = Vec::new();
         for row in rows.into_iter().take(limit as usize) {
-            let Some(timestamp) = parse_flux_time(&row) else { continue; };
-            let Some(id) = row.get("device_id").filter(|v| !v.is_empty()) else { continue; };
+            let Some(timestamp) = parse_flux_time(&row) else {
+                continue;
+            };
+            let Some(id) = row.get("device_id").filter(|v| !v.is_empty()) else {
+                continue;
+            };
             out.push(Reading {
                 id: 0,
                 device_id: id.clone(),
@@ -43,15 +71,40 @@ pub async fn query_zones(
     limit: u32,
 ) -> Result<Vec<ZoneReading>> {
     if settings.version == "1" {
-        query_zones_v1(client, settings, zone_id, start, stop, bucket_seconds, limit).await
+        query_zones_v1(
+            client,
+            settings,
+            zone_id,
+            start,
+            stop,
+            bucket_seconds,
+            limit,
+        )
+        .await
     } else {
-        let tags = if let Some(value) = zone_id { format!(" |> filter(fn: (r) => r.zone_id == {})", flux_string(value)) } else { String::new() };
-        let query = flux_query(settings, ZONE_MEASUREMENT, &tags, &["zone_id", "device_id"], start, stop, bucket_seconds);
+        let tags = if let Some(value) = zone_id {
+            format!(" |> filter(fn: (r) => r.zone_id == {})", flux_string(value))
+        } else {
+            String::new()
+        };
+        let query = flux_query(
+            settings,
+            ZONE_MEASUREMENT,
+            &tags,
+            &["zone_id", "device_id"],
+            start,
+            stop,
+            bucket_seconds,
+        );
         let rows = query_v2(client, settings, &query).await?;
         let mut out = Vec::new();
         for row in rows.into_iter().take(limit as usize) {
-            let Some(timestamp) = parse_flux_time(&row) else { continue; };
-            let Some(zone) = row.get("zone_id").filter(|v| !v.is_empty()) else { continue; };
+            let Some(timestamp) = parse_flux_time(&row) else {
+                continue;
+            };
+            let Some(zone) = row.get("zone_id").filter(|v| !v.is_empty()) else {
+                continue;
+            };
             out.push(ZoneReading {
                 id: 0,
                 zone_id: zone.clone(),
@@ -65,7 +118,10 @@ pub async fn query_zones(
                 outdoor_temperature: row_f64(&row, "outdoor_temperature"),
                 power: row_f64(&row, "power").unwrap_or(0.0) >= 0.5,
                 mode: "history".into(),
-                fan_speed: row_f64(&row, "fan_speed").unwrap_or(0.0).round().clamp(0.0, 5.0) as u8,
+                fan_speed: row_f64(&row, "fan_speed")
+                    .unwrap_or(0.0)
+                    .round()
+                    .clamp(0.0, 5.0) as u8,
                 demand: row_f64(&row, "demand").unwrap_or(0.0) >= 0.5,
                 control_source: "influx".into(),
                 active_preset: "history".into(),
@@ -86,16 +142,46 @@ pub async fn query_ha(
     limit: u32,
 ) -> Result<Vec<HaReading>> {
     if settings.version == "1" {
-        query_ha_v1(client, settings, entity_id, start, stop, bucket_seconds, limit).await
+        query_ha_v1(
+            client,
+            settings,
+            entity_id,
+            start,
+            stop,
+            bucket_seconds,
+            limit,
+        )
+        .await
     } else {
-        let tags = if let Some(value) = entity_id { format!(" |> filter(fn: (r) => r.entity_id == {})", flux_string(value)) } else { String::new() };
-        let query = flux_query(settings, HA_MEASUREMENT, &tags, &["entity_id", "zone_id", "kind"], start, stop, bucket_seconds);
+        let tags = if let Some(value) = entity_id {
+            format!(
+                " |> filter(fn: (r) => r.entity_id == {})",
+                flux_string(value)
+            )
+        } else {
+            String::new()
+        };
+        let query = flux_query(
+            settings,
+            HA_MEASUREMENT,
+            &tags,
+            &["entity_id", "zone_id", "kind"],
+            start,
+            stop,
+            bucket_seconds,
+        );
         let rows = query_v2(client, settings, &query).await?;
         let mut out = Vec::new();
         for row in rows.into_iter().take(limit as usize) {
-            let Some(timestamp) = parse_flux_time(&row) else { continue; };
-            let Some(entity) = row.get("entity_id").filter(|v| !v.is_empty()) else { continue; };
-            let Some(temperature) = row_f64(&row, "temperature") else { continue; };
+            let Some(timestamp) = parse_flux_time(&row) else {
+                continue;
+            };
+            let Some(entity) = row.get("entity_id").filter(|v| !v.is_empty()) else {
+                continue;
+            };
+            let Some(temperature) = row_f64(&row, "temperature") else {
+                continue;
+            };
             out.push(HaReading {
                 id: 0,
                 entity_id: entity.clone(),
@@ -110,23 +196,56 @@ pub async fn query_ha(
     }
 }
 
-async fn query_devices_v1(client: &Client, settings: &InfluxDbSettings, device_id: Option<&str>, start: DateTime<Utc>, stop: DateTime<Utc>, bucket: i64, limit: u32) -> Result<Vec<Reading>> {
-    let filter = device_id.map(|id| format!(" AND \"device_id\"='{}'", influxql_string(id))).unwrap_or_default();
+async fn query_devices_v1(
+    client: &Client,
+    settings: &InfluxDbSettings,
+    device_id: Option<&str>,
+    start: DateTime<Utc>,
+    stop: DateTime<Utc>,
+    bucket: i64,
+    limit: u32,
+) -> Result<Vec<Reading>> {
+    let filter = device_id
+        .map(|id| format!(" AND \"device_id\"='{}'", influxql_string(id)))
+        .unwrap_or_default();
     let q = format!("SELECT mean(\"indoor_temperature\") AS \"indoor_temperature\",mean(\"outdoor_temperature\") AS \"outdoor_temperature\",mean(\"target_temperature\") AS \"target_temperature\",max(\"power\") AS \"power\" FROM \"{DEVICE_MEASUREMENT}\" WHERE time >= '{}' AND time < '{}'{} GROUP BY time({}s),\"device_id\" fill(none) LIMIT {}", start.to_rfc3339(), stop.to_rfc3339(), filter, bucket.max(1), limit);
     let series = query_v1(client, settings, &q).await?;
     let mut out = Vec::new();
     for item in series {
         let device = item.tags.get("device_id").cloned().unwrap_or_default();
         for row in item.rows {
-            let Some(timestamp) = row_time(&row) else { continue; };
-            out.push(Reading { id:0, device_id:device.clone(), timestamp, indoor_temperature:row_num(&row,"indoor_temperature"), outdoor_temperature:row_num(&row,"outdoor_temperature"), target_temperature:row_num(&row,"target_temperature").unwrap_or(0.0), power:row_num(&row,"power").unwrap_or(0.0)>=0.5, source:"influx".into() });
+            let Some(timestamp) = row_time(&row) else {
+                continue;
+            };
+            out.push(Reading {
+                id: 0,
+                device_id: device.clone(),
+                timestamp,
+                indoor_temperature: row_num(&row, "indoor_temperature"),
+                outdoor_temperature: row_num(&row, "outdoor_temperature"),
+                target_temperature: row_num(&row, "target_temperature").unwrap_or(0.0),
+                power: row_num(&row, "power").unwrap_or(0.0) >= 0.5,
+                source: "influx".into(),
+            });
         }
     }
-    out.sort_by_key(|row| row.timestamp); out.truncate(limit as usize); Ok(out)
+    out.sort_by_key(|row| row.timestamp);
+    out.truncate(limit as usize);
+    Ok(out)
 }
 
-async fn query_zones_v1(client: &Client, settings: &InfluxDbSettings, zone_id: Option<&str>, start: DateTime<Utc>, stop: DateTime<Utc>, bucket: i64, limit: u32) -> Result<Vec<ZoneReading>> {
-    let filter = zone_id.map(|id| format!(" AND \"zone_id\"='{}'", influxql_string(id))).unwrap_or_default();
+async fn query_zones_v1(
+    client: &Client,
+    settings: &InfluxDbSettings,
+    zone_id: Option<&str>,
+    start: DateTime<Utc>,
+    stop: DateTime<Utc>,
+    bucket: i64,
+    limit: u32,
+) -> Result<Vec<ZoneReading>> {
+    let filter = zone_id
+        .map(|id| format!(" AND \"zone_id\"='{}'", influxql_string(id)))
+        .unwrap_or_default();
     let q = format!("SELECT mean(\"gree_temperature\") AS \"gree_temperature\",mean(\"external_temperature\") AS \"external_temperature\",mean(\"control_temperature\") AS \"control_temperature\",mean(\"target_temperature\") AS \"target_temperature\",mean(\"device_setpoint\") AS \"device_setpoint\",mean(\"outdoor_temperature\") AS \"outdoor_temperature\",max(\"power\") AS \"power\",mean(\"fan_speed\") AS \"fan_speed\",max(\"demand\") AS \"demand\" FROM \"{ZONE_MEASUREMENT}\" WHERE time >= '{}' AND time < '{}'{} GROUP BY time({}s),\"zone_id\",\"device_id\" fill(none) LIMIT {}", start.to_rfc3339(), stop.to_rfc3339(), filter, bucket.max(1), limit);
     let series = query_v1(client, settings, &q).await?;
     let mut out = Vec::new();
@@ -134,81 +253,210 @@ async fn query_zones_v1(client: &Client, settings: &InfluxDbSettings, zone_id: O
         let zone = item.tags.get("zone_id").cloned().unwrap_or_default();
         let device = item.tags.get("device_id").cloned().unwrap_or_default();
         for row in item.rows {
-            let Some(timestamp) = row_time(&row) else { continue; };
-            out.push(ZoneReading { id:0, zone_id:zone.clone(), device_id:device.clone(), timestamp, gree_temperature:row_num(&row,"gree_temperature"), external_temperature:row_num(&row,"external_temperature"), control_temperature:row_num(&row,"control_temperature"), target_temperature:row_num(&row,"target_temperature"), device_setpoint:row_num(&row,"device_setpoint"), outdoor_temperature:row_num(&row,"outdoor_temperature"), power:row_num(&row,"power").unwrap_or(0.0)>=0.5, mode:"history".into(), fan_speed:row_num(&row,"fan_speed").unwrap_or(0.0).round().clamp(0.0,5.0) as u8, demand:row_num(&row,"demand").unwrap_or(0.0)>=0.5, control_source:"influx".into(), active_preset:"history".into() });
+            let Some(timestamp) = row_time(&row) else {
+                continue;
+            };
+            out.push(ZoneReading {
+                id: 0,
+                zone_id: zone.clone(),
+                device_id: device.clone(),
+                timestamp,
+                gree_temperature: row_num(&row, "gree_temperature"),
+                external_temperature: row_num(&row, "external_temperature"),
+                control_temperature: row_num(&row, "control_temperature"),
+                target_temperature: row_num(&row, "target_temperature"),
+                device_setpoint: row_num(&row, "device_setpoint"),
+                outdoor_temperature: row_num(&row, "outdoor_temperature"),
+                power: row_num(&row, "power").unwrap_or(0.0) >= 0.5,
+                mode: "history".into(),
+                fan_speed: row_num(&row, "fan_speed")
+                    .unwrap_or(0.0)
+                    .round()
+                    .clamp(0.0, 5.0) as u8,
+                demand: row_num(&row, "demand").unwrap_or(0.0) >= 0.5,
+                control_source: "influx".into(),
+                active_preset: "history".into(),
+            });
         }
     }
-    out.sort_by_key(|row| row.timestamp); out.truncate(limit as usize); Ok(out)
+    out.sort_by_key(|row| row.timestamp);
+    out.truncate(limit as usize);
+    Ok(out)
 }
 
-async fn query_ha_v1(client: &Client, settings: &InfluxDbSettings, entity_id: Option<&str>, start: DateTime<Utc>, stop: DateTime<Utc>, bucket: i64, limit: u32) -> Result<Vec<HaReading>> {
-    let filter = entity_id.map(|id| format!(" AND \"entity_id\"='{}'", influxql_string(id))).unwrap_or_default();
+async fn query_ha_v1(
+    client: &Client,
+    settings: &InfluxDbSettings,
+    entity_id: Option<&str>,
+    start: DateTime<Utc>,
+    stop: DateTime<Utc>,
+    bucket: i64,
+    limit: u32,
+) -> Result<Vec<HaReading>> {
+    let filter = entity_id
+        .map(|id| format!(" AND \"entity_id\"='{}'", influxql_string(id)))
+        .unwrap_or_default();
     let q = format!("SELECT mean(\"temperature\") AS \"temperature\" FROM \"{HA_MEASUREMENT}\" WHERE time >= '{}' AND time < '{}'{} GROUP BY time({}s),\"entity_id\",\"zone_id\",\"kind\" fill(none) LIMIT {}", start.to_rfc3339(), stop.to_rfc3339(), filter, bucket.max(1), limit);
     let series = query_v1(client, settings, &q).await?;
     let mut out = Vec::new();
     for item in series {
         let entity = item.tags.get("entity_id").cloned().unwrap_or_default();
         let zone = item.tags.get("zone_id").filter(|v| !v.is_empty()).cloned();
-        let kind = item.tags.get("kind").cloned().unwrap_or_else(|| "room".into());
+        let kind = item
+            .tags
+            .get("kind")
+            .cloned()
+            .unwrap_or_else(|| "room".into());
         for row in item.rows {
-            let Some(timestamp) = row_time(&row) else { continue; };
-            let Some(temperature) = row_num(&row,"temperature") else { continue; };
-            out.push(HaReading { id:0, entity_id:entity.clone(), zone_id:zone.clone(), kind:kind.clone(), timestamp, temperature });
+            let Some(timestamp) = row_time(&row) else {
+                continue;
+            };
+            let Some(temperature) = row_num(&row, "temperature") else {
+                continue;
+            };
+            out.push(HaReading {
+                id: 0,
+                entity_id: entity.clone(),
+                zone_id: zone.clone(),
+                kind: kind.clone(),
+                timestamp,
+                temperature,
+            });
         }
     }
-    out.sort_by_key(|row| row.timestamp); out.truncate(limit as usize); Ok(out)
+    out.sort_by_key(|row| row.timestamp);
+    out.truncate(limit as usize);
+    Ok(out)
 }
 
-struct V1Series { tags: HashMap<String,String>, rows: Vec<HashMap<String,Value>> }
+struct V1Series {
+    tags: HashMap<String, String>,
+    rows: Vec<HashMap<String, Value>>,
+}
 
 async fn query_v1(client: &Client, settings: &InfluxDbSettings, q: &str) -> Result<Vec<V1Series>> {
     validate(settings)?;
     let base = settings.url.trim_end_matches('/');
-    let request = client.get(format!("{base}/query")).query(&[("db", settings.database.as_str()), ("q", q)]);
-    let request = if settings.username.trim().is_empty() { request } else { request.basic_auth(&settings.username, Some(&settings.password)) };
+    let request = client
+        .get(format!("{base}/query"))
+        .query(&[("db", settings.database.as_str()), ("q", q)]);
+    let request = if settings.username.trim().is_empty() {
+        request
+    } else {
+        request.basic_auth(&settings.username, Some(&settings.password))
+    };
     let response = request.send().await.context("InfluxDB 1.x query failed")?;
     let status = response.status();
-    let body: Value = response.json().await.context("invalid InfluxDB 1.x JSON response")?;
-    if !status.is_success() { bail!("InfluxDB 1.x query failed ({status}): {body}"); }
-    if let Some(error) = body.pointer("/results/0/error").and_then(Value::as_str) { bail!("InfluxDB 1.x query error: {error}"); }
+    let body: Value = response
+        .json()
+        .await
+        .context("invalid InfluxDB 1.x JSON response")?;
+    if !status.is_success() {
+        bail!("InfluxDB 1.x query failed ({status}): {body}");
+    }
+    if let Some(error) = body.pointer("/results/0/error").and_then(Value::as_str) {
+        bail!("InfluxDB 1.x query error: {error}");
+    }
     let mut out = Vec::new();
-    for series in body.pointer("/results/0/series").and_then(Value::as_array).into_iter().flatten() {
-        let columns: Vec<String> = series.get("columns").and_then(Value::as_array).into_iter().flatten().filter_map(Value::as_str).map(str::to_owned).collect();
-        let tags = series.get("tags").and_then(Value::as_object).map(|map| map.iter().map(|(k,v)|(k.clone(),v.as_str().unwrap_or_default().to_string())).collect()).unwrap_or_default();
+    for series in body
+        .pointer("/results/0/series")
+        .and_then(Value::as_array)
+        .into_iter()
+        .flatten()
+    {
+        let columns: Vec<String> = series
+            .get("columns")
+            .and_then(Value::as_array)
+            .into_iter()
+            .flatten()
+            .filter_map(Value::as_str)
+            .map(str::to_owned)
+            .collect();
+        let tags = series
+            .get("tags")
+            .and_then(Value::as_object)
+            .map(|map| {
+                map.iter()
+                    .map(|(k, v)| (k.clone(), v.as_str().unwrap_or_default().to_string()))
+                    .collect()
+            })
+            .unwrap_or_default();
         let mut rows = Vec::new();
-        for values in series.get("values").and_then(Value::as_array).into_iter().flatten() {
-            let Some(values) = values.as_array() else { continue; };
-            rows.push(columns.iter().cloned().zip(values.iter().cloned()).collect());
+        for values in series
+            .get("values")
+            .and_then(Value::as_array)
+            .into_iter()
+            .flatten()
+        {
+            let Some(values) = values.as_array() else {
+                continue;
+            };
+            rows.push(
+                columns
+                    .iter()
+                    .cloned()
+                    .zip(values.iter().cloned())
+                    .collect(),
+            );
         }
         out.push(V1Series { tags, rows });
     }
     Ok(out)
 }
 
-async fn query_v2(client: &Client, settings: &InfluxDbSettings, query: &str) -> Result<Vec<HashMap<String,String>>> {
+async fn query_v2(
+    client: &Client,
+    settings: &InfluxDbSettings,
+    query: &str,
+) -> Result<Vec<HashMap<String, String>>> {
     validate(settings)?;
     let base = settings.url.trim_end_matches('/');
-    let response = client.post(format!("{base}/api/v2/query"))
+    let response = client
+        .post(format!("{base}/api/v2/query"))
         .query(&[("org", settings.org.as_str())])
         .bearer_auth(settings.token.trim())
         .header(reqwest::header::ACCEPT, "application/csv")
         .header(reqwest::header::CONTENT_TYPE, "application/vnd.flux")
         .body(query.to_string())
-        .send().await.context("InfluxDB 2.x query failed")?;
+        .send()
+        .await
+        .context("InfluxDB 2.x query failed")?;
     let status = response.status();
-    let body = response.text().await.context("cannot read InfluxDB 2.x response")?;
-    if !status.is_success() { bail!("InfluxDB 2.x query failed ({status}): {}", truncate(&body, 500)); }
+    let body = response
+        .text()
+        .await
+        .context("cannot read InfluxDB 2.x response")?;
+    if !status.is_success() {
+        bail!(
+            "InfluxDB 2.x query failed ({status}): {}",
+            truncate(&body, 500)
+        );
+    }
     let mut headers: Option<Vec<String>> = None;
     let mut rows = Vec::new();
-    for line in body.lines().filter(|line| !line.starts_with('#') && !line.trim().is_empty()) {
+    for line in body
+        .lines()
+        .filter(|line| !line.starts_with('#') && !line.trim().is_empty())
+    {
         let record = parse_csv_line(line);
         if headers.is_none() {
             headers = Some(record);
             continue;
         }
-        let row: HashMap<String,String> = headers.as_ref().unwrap().iter().cloned().zip(record.into_iter()).collect();
-        if row.get("_time").map(|value| !value.is_empty()).unwrap_or(false) { rows.push(row); }
+        let row: HashMap<String, String> = headers
+            .as_ref()
+            .unwrap()
+            .iter()
+            .cloned()
+            .zip(record.into_iter())
+            .collect();
+        if row
+            .get("_time")
+            .map(|value| !value.is_empty())
+            .unwrap_or(false)
+        {
+            rows.push(row);
+        }
     }
     Ok(rows)
 }
-

src/influxdb/write.rs

@@ -1,57 +1,129 @@
 pub fn validate(settings: &InfluxDbSettings) -> Result<()> {
-    if !settings.enabled { return Ok(()); }
-    if !matches!(settings.version.as_str(), "1" | "2") { bail!("InfluxDB version must be 1 or 2"); }
+    if !settings.enabled {
+        return Ok(());
+    }
+    if !matches!(settings.version.as_str(), "1" | "2") {
+        bail!("InfluxDB version must be 1 or 2");
+    }
     let parsed = url::Url::parse(settings.url.trim()).context("invalid InfluxDB URL")?;
-    if !matches!(parsed.scheme(), "http" | "https") { bail!("InfluxDB URL must use http or https"); }
-    if settings.version == "1" && settings.database.trim().is_empty() { bail!("InfluxDB 1.x database is required"); }
+    if !matches!(parsed.scheme(), "http" | "https") {
+        bail!("InfluxDB URL must use http or https");
+    }
+    if settings.version == "1" && settings.database.trim().is_empty() {
+        bail!("InfluxDB 1.x database is required");
+    }
     if settings.version == "2" {
-        if settings.org.trim().is_empty() { bail!("InfluxDB 2.x organization is required"); }
-        if settings.bucket.trim().is_empty() { bail!("InfluxDB 2.x bucket is required"); }
-        if settings.token.trim().is_empty() { bail!("InfluxDB 2.x token is required"); }
+        if settings.org.trim().is_empty() {
+            bail!("InfluxDB 2.x organization is required");
+        }
+        if settings.bucket.trim().is_empty() {
+            bail!("InfluxDB 2.x bucket is required");
+        }
+        if settings.token.trim().is_empty() {
+            bail!("InfluxDB 2.x token is required");
+        }
     }
     Ok(())
 }
 
-pub async fn write_device(client: &Client, settings: &InfluxDbSettings, reading: &Reading) -> Result<()> {
-    if !settings.enabled { return Ok(()); }
+pub async fn write_device(
+    client: &Client,
+    settings: &InfluxDbSettings,
+    reading: &Reading,
+) -> Result<()> {
+    if !settings.enabled {
+        return Ok(());
+    }
     let mut fields = Vec::new();
-    push_float(&mut fields, "indoor_temperature", reading.indoor_temperature);
-    push_float(&mut fields, "outdoor_temperature", reading.outdoor_temperature);
-    push_float(&mut fields, "target_temperature", Some(reading.target_temperature));
+    push_float(
+        &mut fields,
+        "indoor_temperature",
+        reading.indoor_temperature,
+    );
+    push_float(
+        &mut fields,
+        "outdoor_temperature",
+        reading.outdoor_temperature,
+    );
+    push_float(
+        &mut fields,
+        "target_temperature",
+        Some(reading.target_temperature),
+    );
     push_int(&mut fields, "power", reading.power as i64);
-    let line = line_protocol(DEVICE_MEASUREMENT, &[("device_id", &reading.device_id)], fields, reading.timestamp)?;
-    write_line(client, settings, line).await
-}
-
-pub async fn write_zone(client: &Client, settings: &InfluxDbSettings, reading: &ZoneReading) -> Result<()> {
-    if !settings.enabled { return Ok(()); }
-    let mut fields = Vec::new();
-    push_float(&mut fields, "gree_temperature", reading.gree_temperature);
-    push_float(&mut fields, "external_temperature", reading.external_temperature);
-    push_float(&mut fields, "control_temperature", reading.control_temperature);
-    push_float(&mut fields, "target_temperature", reading.target_temperature);
-    push_float(&mut fields, "device_setpoint", reading.device_setpoint);
-    push_float(&mut fields, "outdoor_temperature", reading.outdoor_temperature);
-    push_int(&mut fields, "power", reading.power as i64);
-    push_int(&mut fields, "fan_speed", reading.fan_speed as i64);
-    push_int(&mut fields, "demand", reading.demand as i64);
     let line = line_protocol(
-        ZONE_MEASUREMENT,
-        &[("zone_id", &reading.zone_id), ("device_id", &reading.device_id)],
+        DEVICE_MEASUREMENT,
+        &[("device_id", &reading.device_id)],
         fields,
         reading.timestamp,
     )?;
     write_line(client, settings, line).await
 }
 
-pub async fn write_ha(client: &Client, settings: &InfluxDbSettings, reading: &HaReading) -> Result<()> {
-    if !settings.enabled { return Ok(()); }
+pub async fn write_zone(
+    client: &Client,
+    settings: &InfluxDbSettings,
+    reading: &ZoneReading,
+) -> Result<()> {
+    if !settings.enabled {
+        return Ok(());
+    }
+    let mut fields = Vec::new();
+    push_float(&mut fields, "gree_temperature", reading.gree_temperature);
+    push_float(
+        &mut fields,
+        "external_temperature",
+        reading.external_temperature,
+    );
+    push_float(
+        &mut fields,
+        "control_temperature",
+        reading.control_temperature,
+    );
+    push_float(
+        &mut fields,
+        "target_temperature",
+        reading.target_temperature,
+    );
+    push_float(&mut fields, "device_setpoint", reading.device_setpoint);
+    push_float(
+        &mut fields,
+        "outdoor_temperature",
+        reading.outdoor_temperature,
+    );
+    push_int(&mut fields, "power", reading.power as i64);
+    push_int(&mut fields, "fan_speed", reading.fan_speed as i64);
+    push_int(&mut fields, "demand", reading.demand as i64);
+    let line = line_protocol(
+        ZONE_MEASUREMENT,
+        &[
+            ("zone_id", &reading.zone_id),
+            ("device_id", &reading.device_id),
+        ],
+        fields,
+        reading.timestamp,
+    )?;
+    write_line(client, settings, line).await
+}
+
+pub async fn write_ha(
+    client: &Client,
+    settings: &InfluxDbSettings,
+    reading: &HaReading,
+) -> Result<()> {
+    if !settings.enabled {
+        return Ok(());
+    }
     let zone = reading.zone_id.as_deref().unwrap_or("");
     let mut fields = Vec::new();
     push_float(&mut fields, "temperature", Some(reading.temperature));
     let line = line_protocol(
         HA_MEASUREMENT,
-        &[("entity_id", &reading.entity_id), ("zone_id", zone), ("kind", &reading.kind)],
+        &[
+            ("entity_id", &reading.entity_id),
+            ("zone_id", zone),
+            ("kind", &reading.kind),
+        ],
         fields,
         reading.timestamp,
     )?;
@@ -65,35 +137,89 @@ pub async fn write_batch(
     zones: &[ZoneReading],
     ha: &[HaReading],
 ) -> Result<()> {
-    if !settings.enabled { return Ok(()); }
+    if !settings.enabled {
+        return Ok(());
+    }
     let mut lines = Vec::with_capacity(devices.len() + zones.len() + ha.len());
     for reading in devices {
         let mut fields = Vec::new();
-        push_float(&mut fields, "indoor_temperature", reading.indoor_temperature);
-        push_float(&mut fields, "outdoor_temperature", reading.outdoor_temperature);
-        push_float(&mut fields, "target_temperature", Some(reading.target_temperature));
+        push_float(
+            &mut fields,
+            "indoor_temperature",
+            reading.indoor_temperature,
+        );
+        push_float(
+            &mut fields,
+            "outdoor_temperature",
+            reading.outdoor_temperature,
+        );
+        push_float(
+            &mut fields,
+            "target_temperature",
+            Some(reading.target_temperature),
+        );
         push_int(&mut fields, "power", reading.power as i64);
-        lines.push(line_protocol(DEVICE_MEASUREMENT, &[("device_id", &reading.device_id)], fields, reading.timestamp)?);
+        lines.push(line_protocol(
+            DEVICE_MEASUREMENT,
+            &[("device_id", &reading.device_id)],
+            fields,
+            reading.timestamp,
+        )?);
     }
     for reading in zones {
         let mut fields = Vec::new();
         push_float(&mut fields, "gree_temperature", reading.gree_temperature);
-        push_float(&mut fields, "external_temperature", reading.external_temperature);
-        push_float(&mut fields, "control_temperature", reading.control_temperature);
-        push_float(&mut fields, "target_temperature", reading.target_temperature);
+        push_float(
+            &mut fields,
+            "external_temperature",
+            reading.external_temperature,
+        );
+        push_float(
+            &mut fields,
+            "control_temperature",
+            reading.control_temperature,
+        );
+        push_float(
+            &mut fields,
+            "target_temperature",
+            reading.target_temperature,
+        );
         push_float(&mut fields, "device_setpoint", reading.device_setpoint);
-        push_float(&mut fields, "outdoor_temperature", reading.outdoor_temperature);
+        push_float(
+            &mut fields,
+            "outdoor_temperature",
+            reading.outdoor_temperature,
+        );
         push_int(&mut fields, "power", reading.power as i64);
         push_int(&mut fields, "fan_speed", reading.fan_speed as i64);
         push_int(&mut fields, "demand", reading.demand as i64);
-        lines.push(line_protocol(ZONE_MEASUREMENT, &[("zone_id", &reading.zone_id), ("device_id", &reading.device_id)], fields, reading.timestamp)?);
+        lines.push(line_protocol(
+            ZONE_MEASUREMENT,
+            &[
+                ("zone_id", &reading.zone_id),
+                ("device_id", &reading.device_id),
+            ],
+            fields,
+            reading.timestamp,
+        )?);
     }
     for reading in ha {
         let mut fields = Vec::new();
         push_float(&mut fields, "temperature", Some(reading.temperature));
-        lines.push(line_protocol(HA_MEASUREMENT, &[("entity_id", &reading.entity_id), ("zone_id", reading.zone_id.as_deref().unwrap_or("")), ("kind", &reading.kind)], fields, reading.timestamp)?);
+        lines.push(line_protocol(
+            HA_MEASUREMENT,
+            &[
+                ("entity_id", &reading.entity_id),
+                ("zone_id", reading.zone_id.as_deref().unwrap_or("")),
+                ("kind", &reading.kind),
+            ],
+            fields,
+            reading.timestamp,
+        )?);
+    }
+    if lines.is_empty() {
+        return Ok(());
     }
-    if lines.is_empty() { return Ok(()); }
     write_lines(client, settings, lines.join("\n")).await
 }
 
@@ -105,19 +231,32 @@ async fn write_lines(client: &Client, settings: &InfluxDbSettings, body: String)
     validate(settings)?;
     let base = settings.url.trim_end_matches('/');
     let request = if settings.version == "1" {
-        let request = client.post(format!("{base}/write"))
+        let request = client
+            .post(format!("{base}/write"))
             .query(&[("db", settings.database.as_str()), ("precision", "ns")])
             .header(reqwest::header::CONTENT_TYPE, "text/plain; charset=utf-8")
             .body(body.clone());
-        if settings.username.trim().is_empty() { request } else { request.basic_auth(&settings.username, Some(&settings.password)) }
+        if settings.username.trim().is_empty() {
+            request
+        } else {
+            request.basic_auth(&settings.username, Some(&settings.password))
+        }
     } else {
-        client.post(format!("{base}/api/v2/write"))
-            .query(&[("org", settings.org.as_str()), ("bucket", settings.bucket.as_str()), ("precision", "ns")])
+        client
+            .post(format!("{base}/api/v2/write"))
+            .query(&[
+                ("org", settings.org.as_str()),
+                ("bucket", settings.bucket.as_str()),
+                ("precision", "ns"),
+            ])
             .bearer_auth(settings.token.trim())
             .header(reqwest::header::CONTENT_TYPE, "text/plain; charset=utf-8")
             .body(body)
     };
-    let response = request.send().await.context("InfluxDB write request failed")?;
+    let response = request
+        .send()
+        .await
+        .context("InfluxDB write request failed")?;
     if !response.status().is_success() {
         let status = response.status();
         let body = response.text().await.unwrap_or_default();
@@ -125,4 +264,3 @@ async fn write_lines(client: &Client, settings: &InfluxDbSettings, body: String)
     }
     Ok(())
 }
-