Lightweight InfluxDB ingestion proxy. It corrects the house power of a PV system with a balcony inverter.
A balcony inverter feeds power into a socket of your house grid. It sits behind the meter of your main system. Your battery system and your roof inverter never see this power.
They compute the house power from what they measure:
HOUSE_POWER = INVERTER_POWER
+ GRID_IMPORT_POWER
- GRID_EXPORT_POWER
+ BATTERY_DISCHARGING_POWER
- BATTERY_CHARGING_POWER
The balcony inverter reduces the grid import. So the result is too low by exactly the power of the balcony inverter. Your dashboard shows a house that uses less energy than it really does.
Ingest sits between your collectors and InfluxDB. It adds the balcony inverter to the sum and replaces the wrong value.
The formula is simple arithmetic. A template sensor in Home Assistant or a script in ioBroker can add the terms. The sum is not the hard part. The alignment in time is.
A template sensor in Home Assistant, or a script in ioBroker, runs when one input changes. It then reads the current value of every other input. That value is the last one that arrived, whatever its age.
The sensors of a PV system do not report at the same rate. A battery system reports every few seconds. A smart plug reports every 30 seconds. A balcony inverter over MQTT reports once a minute or slower. So the sum adds one fresh value to values that are up to a minute old.
Under a load that changes fast, this error is larger than the correction you want. A kettle that starts between two reports of the balcony inverter moves the result by 2000 W.
Both platforms do keep a history. Home Assistant has the recorder, with the sql integration and the recorder.get_statistics action on top of it. ioBroker has the history adapters. But each of them gives you one query per sensor, on its own polling interval. None of them gives you the aligned value of eight sensors at one exact moment.
Ingest stores every sample with its own timestamp. For the moment of the trigger, it interpolates each term between the two samples that surround that moment. Every term of the sum then belongs to the same moment.
This is the difference that no amount of work removes. Home Assistant and ioBroker can read their history, but they cannot add a value to it. Every result of a calculation enters as a state change now, and the export to InfluxDB follows that state change.
If a collector delivers a batch of older points, the house power for those moments is never computed. This happens after a restart of a collector, and after every network problem. The gap stays in your data.
Ingest works on the timestamp of each line, not on the present. A batch of 5000 old points gets 5000 corrected values, each one at its own timestamp.
Ingest also refuses to give you a wrong number. Every term of the formula needs a sample that is less than 15 minutes old. If one term is too old, Ingest writes no house power at all, and the stats page names the sensor that caused the skip.
You can do this in Home Assistant with states.sensor.balcony_power.last_updated, and in ioBroker with the ts field of getState(). Both are one line. But you have to think of it first, and you have to think of it for all eight terms.
Interpolation, staleness detection, and a write path that carries its own timestamps are all possible on both platforms. It is a project, not a template sensor. Ingest is that project, and it costs you one URL change in your collectors.
Use Ingest when both of these are true:
- You have a balcony inverter, or another producer that your main system cannot measure.
- Your house power values are too low.
If you have one inverter only, your house power is already correct. Ingest then adds nothing, except load on your Raspberry Pi.
HELIOS configures your SOLECTRUS installation. It adds Ingest to your Docker Compose file and sets every environment variable. You do not install or configure Ingest yourself.
HELIOS adds Ingest only when both of these are true:
- Your configuration has at least one balcony sensor.
- Every sensor of the formula comes from a collector that HELIOS manages.
If one input arrives from an external source, HELIOS leaves Ingest out. Ingest then sees a part of the data only, and a house power from partial data is worse than no correction at all. In this case the external source must deliver a correct house power itself.
This is why the SOLECTRUS integrations for Home Assistant and ioBroker do not use Ingest. They write to InfluxDB directly, and you compute the house power on that side.
graph LR
CollectorA[SENEC-Collector]
CollectorB[Shelly-Collector]
CollectorC[MQTT-Collector]
Influx[InfluxDB]
Dashboard[Dashboard]
CollectorA -->|push| Influx
CollectorB -->|push| Influx
CollectorC -->|push| Influx
Influx -->|pull| Dashboard
graph LR
CollectorA[SENEC-Collector]
CollectorB[Shelly-Collector]
CollectorC[MQTT-Collector]
Influx[InfluxDB]
Ingest[Ingest]
Dashboard[Dashboard]
CollectorA -->|push| Ingest
CollectorB -->|push| Ingest
CollectorC -->|push| Ingest
Ingest -->|push| Influx
Influx -->|pull| Dashboard
Ingest accepts InfluxDB v2 Line Protocol over HTTP, so a collector needs a new destination URL and nothing else. It forwards every line. If one line does not parse, Ingest skips that line and forwards the rest of the request.
Ingest recalculates the house power with this formula:
HOUSE_POWER = INVERTER_POWER (total, including balcony inverter)
+ GRID_IMPORT_POWER
+ BATTERY_DISCHARGING_POWER
- BATTERY_CHARGING_POWER
- GRID_EXPORT_POWER
- WALLBOX_POWER
- HEATPUMP_POWER
Whenever one of these sensors updates, Ingest recalculates the house power for the timestamp of that update.
The relevant sensor values do not arrive at the same moment. For each term, Ingest finds the two samples that surround the target timestamp and interpolates between them. If no later sample exists yet, Ingest takes the last one, but only when it is less than 15 minutes old.
All configured sensors must contribute a value. If one of them is older than 15 minutes, Ingest skips the recalculation and writes no house power point.
The formula needs every term. If a term is missing, the result is wrong. Ingest does not replace the missing term with a guess, so it writes nothing at all. A gap in the dashboard is better than a wrong value in the database.
The stats page shows these skips as Skipped (value missing), plus a Missing values by sensor list that names the sensors which caused them.
The Configured sensors list gives a second view. While a sensor delivers, the list shows its rate. If a sensor sends nothing for more than 15 minutes, the list shows the age of its last line instead. This is the same limit that the formula uses, so the list turns red at the moment when the sensor stops to contribute. The buffer keeps the lines of a stopped collector for the whole retention period, so the count and the rate stay unchanged.
So every configured sensor must send data continuously, even when its true value is zero. There are two different causes for a gap:
- The sensor is broken or offline and sends nothing at all. Then the gap is correct. Ingest cannot know whether the true value is zero, so it shows no value instead of a wrong one. Repair the sensor or the collector.
- The sensor still sends, but without a value. This is a configuration problem, and you can fix it in the collector. For the MQTT-Collector, set
MAPPING_X_NULL_TO_ZERO=trueto write a0when the payload holds no value. A balcony inverter that reportsnullat night is a typical example.
The calculated value replaces the original one. If you prefer to store the original value separately, you can define INFLUX_SENSOR_HOUSE_POWER_CALCULATED to write the result to a different measurement and/or field.
Ingest keeps every incoming line in SQLite, for 12 hours by default. This buffer is not a feature of its own. The interpolation needs the samples around a timestamp, so Ingest must hold them.
The buffer has two useful side effects:
- If InfluxDB is unreachable, no data is lost. Ingest keeps the lines and retries with the next batch.
- The stats page shows the rate of every sensor, so you find a collector that stopped.
The database holds two sets of rows, and they have different lifetimes.
The incoming rows hold every value that the collectors sent. The interpolation reads them, and a background worker removes an incoming row that is older than RETENTION_HOURS.
The outgoing rows hold every line that still waits for InfluxDB. Another background worker sends them in batches and removes each line that arrives. RETENTION_HOURS does not touch the outgoing rows, and their number has no limit. If InfluxDB stays unreachable, the database grows until InfluxDB accepts the lines again. The stats page shows the two counts as Values buffered and Lines waiting.
services:
ingest:
image: ghcr.io/solectrus/ingest:latest
environment:
- INFLUX_SENSOR_INVERTER_POWER
- INFLUX_SENSOR_INVERTER_POWER_1
- INFLUX_SENSOR_INVERTER_POWER_2
- INFLUX_SENSOR_INVERTER_POWER_3
- INFLUX_SENSOR_INVERTER_POWER_4
- INFLUX_SENSOR_INVERTER_POWER_5
- INFLUX_SENSOR_GRID_IMPORT_POWER
- INFLUX_SENSOR_GRID_EXPORT_POWER
- INFLUX_SENSOR_BATTERY_DISCHARGING_POWER
- INFLUX_SENSOR_BATTERY_CHARGING_POWER
- INFLUX_SENSOR_WALLBOX_POWER
- INFLUX_SENSOR_HEATPUMP_POWER
- INFLUX_SENSOR_HOUSE_POWER
- INFLUX_EXCLUDE_FROM_HOUSE_POWER
- INFLUX_SENSOR_HOUSE_POWER_CALCULATED
- INFLUX_HOST
- INFLUX_PORT
- INFLUX_SCHEMA
- STATS_PASSWORD
depends_on:
- influxdb
ports:
- 4567:4567
volumes:
- # Just an example!
- # Change to a valid path on your system where Ingest can store its SQLite database
- ./path/to/ingest-data:/app/data
influxdb:
image: influxdb:2.9-alpine
ports:
- 8086:8086
volumes:
- # Just an example
- ./path/to/influx-data:/var/lib/influxdb2Define measurement and field names for each sensor. Format: measurement:field, for example SENEC:inverter_power. Leave a sensor empty when it is not available.
| Variable | Description |
|---|---|
INFLUX_SENSOR_INVERTER_POWER |
Inverter power (total) |
INFLUX_SENSOR_INVERTER_POWER_1 |
Inverter power (1) |
INFLUX_SENSOR_INVERTER_POWER_2 |
Inverter power (2) |
INFLUX_SENSOR_INVERTER_POWER_3 |
Inverter power (3) |
INFLUX_SENSOR_INVERTER_POWER_4 |
Inverter power (4) |
INFLUX_SENSOR_INVERTER_POWER_5 |
Inverter power (5) |
INFLUX_SENSOR_GRID_IMPORT_POWER |
Grid import power |
INFLUX_SENSOR_GRID_EXPORT_POWER |
Grid export power |
INFLUX_SENSOR_BATTERY_DISCHARGING_POWER |
Battery discharging power |
INFLUX_SENSOR_BATTERY_CHARGING_POWER |
Battery charging power |
INFLUX_SENSOR_WALLBOX_POWER |
Wallbox power |
INFLUX_SENSOR_HEATPUMP_POWER |
Heat pump power |
INFLUX_SENSOR_HOUSE_POWER |
House power |
The inverter power is calculated as follows:
- If
INFLUX_SENSOR_INVERTER_POWERpower is given:
Total inverter power = INFLUX_SENSOR_INVERTER_POWER
- Otherwise, if
INFLUX_SENSOR_INVERTER_POWERis not given:
Total inverter power = INFLUX_SENSOR_INVERTER_POWER_1 +
INFLUX_SENSOR_INVERTER_POWER_2 +
INFLUX_SENSOR_INVERTER_POWER_3 +
INFLUX_SENSOR_INVERTER_POWER_4 +
INFLUX_SENSOR_INVERTER_POWER_5
| Variable | Description | Note |
|---|---|---|
INFLUX_EXCLUDE_FROM_HOUSE_POWER |
Exclude specific sensors from house power | Optional |
INFLUX_SENSOR_HOUSE_POWER_CALCULATED |
Output for calculated house power | Optional |
INFLUX_HOST |
InfluxDB host, for example influxdb |
Required |
INFLUX_PORT |
InfluxDB port | Default: 8086 |
INFLUX_SCHEMA |
InfluxDB schema | Default: http |
RETENTION_HOURS |
How long Ingest keeps the incoming rows | Default: 12 |
STATS_PASSWORD |
Password for stats endpoint | Optional |
Stores and forwards incoming Line Protocol data to InfluxDB. Triggers recalculation of house power if relevant.
Shows a stats page (requires a password if configured), with throughput, queue size, buffer status, and more.
The page has three tabs. Each tab has a path of its own, so the page keeps the tab when it reloads itself every 30 seconds. A dot beside the name of a tab says that the tab needs attention.
| Tab | Path | What it shows |
|---|---|---|
Pipeline |
/ |
What arrives, what the buffer holds, and what reaches InfluxDB |
Sensors |
/sensors |
Every configured sensor, and every other stream that Ingest forwards |
House power |
/house-power |
The formula with the values of the last calculation |
The tab prints the formula term by term, with the value that each sensor contributed and the result of the sum. The terms come from the calculation itself, so what the page shows is what Ingest added up. A bar beside each term gives its share of the largest one.
Below the sum stand two more rows: the house power that the collector delivered, and the difference between the two. Ingest drops the delivered value and writes its own in its place, so this difference is the correction that the service makes.
The page shows a second calculation below the first one when INFLUX_EXCLUDE_FROM_HOUSE_POWER names a sensor. SOLECTRUS subtracts every excluded sensor from the house power again before it draws the dashboard. So Ingest writes one value, and a reader sees another one. The second block names both.
While a stream delivers, the page shows its rate. If a stream sends nothing for more than 15 minutes, the page shows the age of its last line instead.
Only a configured sensor gets a colour for this, because SOLECTRUS needs its value. In the Other incoming data list the age is a fact and not a fault. A collector can send slower on purpose, and Ingest does not know how often such a stream arrives.
Returns JSON with HTTP 200 if the service is running (useful for detailed health checks).
Returns just HTTP 204 if the service is running (useful for health checks).
curl -X POST "http://localhost:4567/api/v2/write?bucket=my-bucket&org=my-org&precision=ns" \
-H "Authorization: Token my-token" \
-H "Content-Type: application/json" \
--data-raw "test_measurement,location=office value=42i $(( $(date +%s) * 1000000000 ))"If anything goes wrong, please look at the logs!
You can. Ingest forwards every line it gets, and it starts a recalculation only when a line carries one of the eight relevant sensors. Everything else passes through untouched. So a collector that sends unrelated data costs you nothing but a row in the buffer.
Ingest needs the eight relevant sensors, so route those collectors through it. For the rest it is your choice. One point of care is a collector that writes a long time series in one request. The Forecast-Collector is the example. Such a series fills the buffer for the whole retention period and gets no benefit from it. Send that collector directly to InfluxDB.
Telegraf is an agent for data collection. Ingest is a proxy that processes and forwards data. So you can reuse your existing collectors and change the destination URL only.
Ingest answers a write as soon as the line is safe in SQLite. The forwarding to InfluxDB runs in the background, in batches. Every line carries its own timestamp, so the moment it reaches InfluxDB changes no data.