InfluxFlux.jl

influx_server(uri, org, api_token) -> InfluxServer

Create a handle to an InfluxDB v2 server.

Arguments

• uri: Base URL of the server, e.g. "http://localhost:8086".
• org: InfluxDB organisation name.
• api_token: Read-access API token.

flux(srv, flux_query) -> Vector{UInt8}

Execute a raw Flux query against srv and return the response body as bytes.

Throws an InfluxFluxError on any non-200 HTTP response.

flux_to_dataframe(srv, flux_query) -> DataFrame

Execute a Flux query and return the result as a single DataFrame.

df = flux_to_dataframe(srv, """
    from(bucket: "sensors")
      |> range(start: -1h)
      |> filter(fn: (r) => r._measurement == "cpu_load")
""")
# df has columns: result, table, _start, _stop, _time, _value, _field, _measurement, host, …

Throws if the query returns more than one result table. Use flux_to_dataframe_multi when multiple results are expected.

flux_to_dataframe_multi(srv, flux_query) -> NamedTuple

Execute a Flux query and return a NamedTuple mapping each named result to a Vector{DataFrame} (one element per table group).

Use this when a query yields multiple results with yield(name: ...):

q = """
    from(bucket: "sensors")
      |> range(start: -1h)
      |> filter(fn: (r) => r._measurement == "temperature")
      |> yield(name: "temp")

    from(bucket: "sensors")
      |> range(start: -1h)
      |> filter(fn: (r) => r._measurement == "humidity")
      |> yield(name: "hum")
"""
result = flux_to_dataframe_multi(srv, q)
result.temp   # Vector{DataFrame} for temperature tables
result.hum    # Vector{DataFrame} for humidity tables

clean_influx_df(df) -> DataFrame

Drop the InfluxDB bookkeeping columns (result, table, _start, _stop, _measurement, Column1) from a DataFrame, leaving only the time and field columns.

raw = flux_to_dataframe(srv, "from(bucket: \"env\") |> range(start: -1h) |> pivot(...)")
# raw has columns: result, table, _start, _stop, _measurement, _time, temp, humidity
df = clean_influx_df(raw)
# df has columns: _time, temp, humidity

measurement(srv, bucket, measurement_name, from, to) -> DataFrame

Fetch a measurement over a time range and return it as a single DataFrame with one column per field and a _time column of epoch-nanosecond integers.

df = measurement(srv, "sensors", "temperature", now() - Hour(1), now())
# df columns: _time, indoor, outdoor

Throws if the query returns more than one table group (i.e. multiple tag-set combinations). Use measurement_multi in that case.

measurement_multi(srv, bucket, measurement_name, from, to) -> Vector{DataFrame}

Fetch a measurement over a time range and return one DataFrame per table group (typically one per tag-set combination). Each DataFrame has one column per field plus a _time column of epoch-nanosecond integers.

dfs = measurement_multi(srv, "sensors", "temperature", now() - Minute(10), now())
# dfs[1] — columns: _time, value  (tag set A)
# dfs[2] — columns: _time, value  (tag set B)

See also measurement when only one table group is expected.

aggregate_measurement(srv, bucket, measurement_name, from, to, window; fn="mean")
-> DataFrame

Downsample a measurement into fixed-width time buckets and return a single DataFrame. See aggregate_measurement_multi for supported fn values.

# Hourly means over the past day
df = aggregate_measurement(srv, "sensors", "temperature",
                           now() - Day(1), now(), Hour(1))
# df columns: _time, indoor, outdoor

# 10-minute maxima
df = aggregate_measurement(srv, "power", "consumption",
                           now() - Hour(6), now(), Minute(10); fn="max")

Throws if the query returns more than one table group. Use aggregate_measurement_multi in that case.

aggregate_measurement_multi(srv, bucket, measurement_name, from, to, window; fn="mean")
-> Vector{DataFrame}

Like measurement_multi but downsamples each field with an aggregate function applied over window-sized time buckets. Returns one DataFrame per tag-set combination.

# 5-minute means over the last hour
dfs = aggregate_measurement_multi(srv, "sensors", "temperature",
                                  now() - Hour(1), now(), Minute(5))

# Maximum instead of mean
dfs = aggregate_measurement_multi(srv, "sensors", "temperature",
                                  now() - Hour(1), now(), Minute(5); fn="max")

window can be any Period, e.g. Second(30), Minute(5), Hour(1).

Common values for fn: "mean", "median", "sum", "count", "min", "max". See the Flux aggregateWindow docs for the full list.

list_buckets(srv) -> Vector{String}

Return the names of all buckets visible to the configured API token.

list_buckets(srv)
# ["_monitoring", "_tasks", "sensors", "power"]

list_measurements(srv, bucket) -> Vector{String}

Return the measurement names stored in bucket.

list_measurements(srv, "sensors")
# ["humidity", "pressure", "temperature"]

list_fields(srv, bucket) -> Vector{String}
list_fields(srv, bucket, measurement) -> Vector{String}

Return field key names in bucket, optionally filtered to a single measurement.

list_fields(srv, "sensors")
# ["humidity", "pressure", "temperature"]

list_fields(srv, "sensors", "temperature")
# ["indoor", "outdoor"]

time_spec_to_epoc_ns(time_spec::TimeSpec) -> Int

Convert a TimeSpec value to an integer epoch-nanosecond timestamp (Unix × 10⁹).

Accepts an Int (returned as-is), a DateTime (treated as UTC), or a ZonedDateTime (converted to UTC first).