v0.5.0
This commit is contained in:
Mateusz Gruszczyński
+403
@@ -0,0 +1,403 @@
|
||||
use anyhow::{anyhow, bail, Context, Result};
|
||||
use chrono::{DateTime, Utc};
|
||||
use reqwest::Client;
|
||||
use serde_json::Value;
|
||||
use std::collections::HashMap;
|
||||
|
||||
use crate::models::{HaReading, InfluxDbSettings, Reading, ZoneReading};
|
||||
|
||||
const DEVICE_MEASUREMENT: &str = "gree_device";
|
||||
const ZONE_MEASUREMENT: &str = "gree_zone";
|
||||
const HA_MEASUREMENT: &str = "gree_ha";
|
||||
|
||||
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"); }
|
||||
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 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"); }
|
||||
}
|
||||
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_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)],
|
||||
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)],
|
||||
fields,
|
||||
reading.timestamp,
|
||||
)?;
|
||||
write_line(client, settings, line).await
|
||||
}
|
||||
|
||||
pub async fn write_batch(
|
||||
client: &Client,
|
||||
settings: &InfluxDbSettings,
|
||||
devices: &[Reading],
|
||||
zones: &[ZoneReading],
|
||||
ha: &[HaReading],
|
||||
) -> Result<()> {
|
||||
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_int(&mut fields, "power", reading.power as i64);
|
||||
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, "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);
|
||||
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)?);
|
||||
}
|
||||
if lines.is_empty() { return Ok(()); }
|
||||
write_lines(client, settings, lines.join("\n")).await
|
||||
}
|
||||
|
||||
async fn write_line(client: &Client, settings: &InfluxDbSettings, line: String) -> Result<()> {
|
||||
write_lines(client, settings, line).await
|
||||
}
|
||||
|
||||
async fn write_lines(client: &Client, settings: &InfluxDbSettings, body: String) -> Result<()> {
|
||||
validate(settings)?;
|
||||
let base = settings.url.trim_end_matches('/');
|
||||
let request = if settings.version == "1" {
|
||||
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)) }
|
||||
} else {
|
||||
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")?;
|
||||
if !response.status().is_success() {
|
||||
let status = response.status();
|
||||
let body = response.text().await.unwrap_or_default();
|
||||
bail!("InfluxDB write failed ({status}): {}", truncate(&body, 300));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn query_devices(
|
||||
client: &Client,
|
||||
settings: &InfluxDbSettings,
|
||||
device_id: Option<&str>,
|
||||
start: DateTime<Utc>,
|
||||
stop: DateTime<Utc>,
|
||||
bucket_seconds: i64,
|
||||
limit: u32,
|
||||
) -> Result<Vec<Reading>> {
|
||||
if settings.version == "1" {
|
||||
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 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; };
|
||||
out.push(Reading {
|
||||
id: 0,
|
||||
device_id: id.clone(),
|
||||
timestamp,
|
||||
indoor_temperature: row_f64(&row, "indoor_temperature"),
|
||||
outdoor_temperature: row_f64(&row, "outdoor_temperature"),
|
||||
target_temperature: row_f64(&row, "target_temperature").unwrap_or(0.0),
|
||||
power: row_f64(&row, "power").unwrap_or(0.0) >= 0.5,
|
||||
source: "influx".into(),
|
||||
});
|
||||
}
|
||||
out.sort_by_key(|row| row.timestamp);
|
||||
Ok(out)
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn query_zones(
|
||||
client: &Client,
|
||||
settings: &InfluxDbSettings,
|
||||
zone_id: Option<&str>,
|
||||
start: DateTime<Utc>,
|
||||
stop: DateTime<Utc>,
|
||||
bucket_seconds: i64,
|
||||
limit: u32,
|
||||
) -> Result<Vec<ZoneReading>> {
|
||||
if settings.version == "1" {
|
||||
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 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; };
|
||||
out.push(ZoneReading {
|
||||
id: 0,
|
||||
zone_id: zone.clone(),
|
||||
device_id: row.get("device_id").cloned().unwrap_or_default(),
|
||||
timestamp,
|
||||
gree_temperature: row_f64(&row, "gree_temperature"),
|
||||
external_temperature: row_f64(&row, "external_temperature"),
|
||||
control_temperature: row_f64(&row, "control_temperature"),
|
||||
target_temperature: row_f64(&row, "target_temperature"),
|
||||
device_setpoint: row_f64(&row, "device_setpoint"),
|
||||
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,
|
||||
demand: row_f64(&row, "demand").unwrap_or(0.0) >= 0.5,
|
||||
control_source: "influx".into(),
|
||||
active_preset: "history".into(),
|
||||
});
|
||||
}
|
||||
out.sort_by_key(|row| row.timestamp);
|
||||
Ok(out)
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn query_ha(
|
||||
client: &Client,
|
||||
settings: &InfluxDbSettings,
|
||||
entity_id: Option<&str>,
|
||||
start: DateTime<Utc>,
|
||||
stop: DateTime<Utc>,
|
||||
bucket_seconds: i64,
|
||||
limit: u32,
|
||||
) -> Result<Vec<HaReading>> {
|
||||
if settings.version == "1" {
|
||||
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 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; };
|
||||
out.push(HaReading {
|
||||
id: 0,
|
||||
entity_id: entity.clone(),
|
||||
zone_id: row.get("zone_id").filter(|v| !v.is_empty()).cloned(),
|
||||
kind: row.get("kind").cloned().unwrap_or_else(|| "room".into()),
|
||||
timestamp,
|
||||
temperature,
|
||||
});
|
||||
}
|
||||
out.sort_by_key(|row| row.timestamp);
|
||||
Ok(out)
|
||||
}
|
||||
}
|
||||
|
||||
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() });
|
||||
}
|
||||
}
|
||||
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();
|
||||
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();
|
||||
for item in series {
|
||||
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() });
|
||||
}
|
||||
}
|
||||
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();
|
||||
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());
|
||||
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 });
|
||||
}
|
||||
}
|
||||
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>> }
|
||||
|
||||
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 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 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();
|
||||
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());
|
||||
}
|
||||
out.push(V1Series { tags, rows });
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
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"))
|
||||
.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")?;
|
||||
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 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()) {
|
||||
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); }
|
||||
}
|
||||
Ok(rows)
|
||||
}
|
||||
|
||||
fn parse_csv_line(line: &str) -> Vec<String> {
|
||||
let mut out = Vec::new();
|
||||
let mut field = String::new();
|
||||
let mut chars = line.chars().peekable();
|
||||
let mut quoted = false;
|
||||
while let Some(ch) = chars.next() {
|
||||
match ch {
|
||||
'"' if quoted && chars.peek() == Some(&'"') => {
|
||||
field.push('"');
|
||||
chars.next();
|
||||
}
|
||||
'"' => quoted = !quoted,
|
||||
',' if !quoted => {
|
||||
out.push(std::mem::take(&mut field));
|
||||
}
|
||||
_ => field.push(ch),
|
||||
}
|
||||
}
|
||||
out.push(field);
|
||||
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(",");
|
||||
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 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)) }
|
||||
Reference in New Issue
Block a user