9 Commits

Author SHA1 Message Date
rmcguire 38078a330e update changelog, bump chart
Build and Publish / go-binaries (push) Failing after 58s
Build and Publish / helm-release (push) Has been skipped
Build and Publish / check-chart (push) Successful in 10s
Build and Publish / container-images (push) Successful in 2m14s
2026-07-05 17:01:01 -04:00
rmcguire 047f7c233d remove rheemcloud pkg, implement against rest api only
Build and Publish / helm-release (push) Has been skipped
Build and Publish / go-binaries (push) Has been skipped
Build and Publish / container-images (push) Has been skipped
Build and Publish / check-chart (push) Successful in 12s
2026-07-05 16:59:53 -04:00
rmcguire dcdf5e4230 bump chart
Build and Publish / go-binaries (push) Successful in 1m41s
Build and Publish / check-chart (push) Successful in 10s
Build and Publish / helm-release (push) Has been skipped
Build and Publish / container-images (push) Successful in 5m19s
2026-07-05 14:56:30 -04:00
rmcguire 1b42d26ffc add changelog 2026-07-05 14:56:10 -04:00
rmcguire a8e0a6ef1e support tls insecure due to cloudblade old TLS cert 2026-07-05 14:53:10 -04:00
rmcguire 059b89b5cd add TODO
Build and Publish / helm-release (push) Has been skipped
Build and Publish / container-images (push) Has been skipped
Build and Publish / go-binaries (push) Has been skipped
Build and Publish / check-chart (push) Successful in 10s
2026-07-04 17:47:54 -04:00
rmcguire 87e53bd620 bump chart versions
Build and Publish / container-images (push) Successful in 2m10s
Build and Publish / helm-release (push) Has been skipped
Build and Publish / check-chart (push) Successful in 16s
Build and Publish / go-binaries (push) Successful in 5m29s
2026-07-04 17:42:50 -04:00
rmcguire c481d3f032 add ca-certs to image
Build and Publish / check-chart (push) Successful in 14s
Build and Publish / helm-release (push) Has been skipped
Build and Publish / go-binaries (push) Successful in 2m35s
Build and Publish / container-images (push) Successful in 3m1s
2026-07-04 17:42:03 -04:00
rmcguire 1918138dbf fix chart app version
Build and Publish / container-images (push) Successful in 2m12s
Build and Publish / helm-release (push) Has been skipped
Build and Publish / check-chart (push) Successful in 17s
Build and Publish / go-binaries (push) Successful in 4m31s
2026-07-04 17:27:34 -04:00
19 changed files with 906 additions and 250 deletions
+31 -13
View File
@@ -9,14 +9,25 @@ Exporter for Rheem EcoNet / Rheemcloud water heaters, built on the
`appServices` slice (registers `&econet.EconetService{}`) and the
`serviceConfig.LoadEnv()` call (see Config gotcha below).
- `pkg/econet/econet.go``EconetService` implements `service.AppService`.
It **owns the single `*rheemcloud.Client`** and shares it with the gRPC,
MCP, and metrics sub-servers. `Init` connects (fail-fast if creds missing);
the returned `ShutdownFunc` stops metrics and closes the client.
It **owns the single `econetclient.Client`** and shares it with the gRPC,
MCP, and metrics sub-servers. `Init` fails fast if creds are missing, then
refreshes device state on a ticker **in a background goroutine** (`pollLoop`,
interval `pollInterval`) so a slow or unreachable cloud never blocks
HTTP/gRPC server startup. The client is stateless REST, so `ShutdownFunc` is
effectively a no-op (the poll goroutine stops on ctx cancel).
- `pkg/econet/econetclient/` — a minimal EcoNet REST client (ClearBlade
backend). `client.go` does the 3 REST calls (auth → `getUserDataForApp`
`dynamicAction` usage) and caches a snapshot under an `RWMutex`; `device.go`
decodes the `@...` datapoints into a flat `Device` struct (ported from
kevinburke/rheemcloud-go, minus MQTT). `Devices()` is empty until the first
`Refresh` succeeds — consumers just see an empty list (no nil-guarding).
**No MQTT**, so `WiFiSignal` and `Running`/`RunningState` stay zero.
- `pkg/econet/econetgrpc/``EconetService` gRPC impl (`ListDevices`,
`GetDevice`) + `deviceToProto`. Does **not** own a client; it's injected.
- `pkg/econet/econetmcp/` — MCP server at `/api/mcp`; the `list_water_heaters`
tool delegates to the gRPC server.
- `pkg/econet/econetmetrics/` — OTEL observable gauges + a usage poller.
- `pkg/econet/econetmetrics/` — OTEL observable gauges (read the client
snapshot at scrape time; no poller of its own).
## Config gotcha (important)
@@ -30,19 +41,26 @@ that's all it takes — no extra wiring.
- `costPerKWH` is **US dollars per kWh** (0.18 = 18¢), feeding
`econet_energy_cost_dollars`.
- `econetTLSInsecure` (env `ECONET_TLS_INSECURE`) skips TLS certificate
verification when connecting to the EcoNet cloud API. It exists because the
upstream host (cloudblade) serves a
DigiCert Global Root G1 chain, which is no longer trusted by default in
current root stores. **Evaluate removing this later** — prefer fixing the
trust chain (e.g. bundling the needed intermediate/root CA) over disabling
verification outright.
## Metrics conventions
- Live state is read at scrape time inside a `RegisterCallback` over
`client.Devices()` (cheap, in-memory, always current — the rheemcloud client
keeps state fresh over MQTT). Usage (`EnergyUsage`/`WaterUsage`) is REST and
slow, so a background poller (`usageInterval`) caches it; the usage callback
reads the cache.
- Live state and usage are both read at scrape time inside `RegisterCallback`s
over `client.Devices()` (cheap, in-memory). The `pollLoop` in `econet.go`
keeps the snapshot fresh over REST (`client.Refresh`), which folds daily
energy/water usage totals into each `Device`. No usage cache or event
draining — the callbacks just read the current snapshot.
- **Attributes**: dotted/namespaced `econet.*` (semantic-convention style),
low-cardinality only (serial, device_id, friendly_name, type, generic_type,
mode, energy_type). **Never** put changing values (timestamps) in attributes
`econet_last_updated_seconds` is a metric *value*, and it's suppressed
until the first MQTT update so it never emits a garbage epoch.
until the first refresh so it never emits a garbage epoch.
- **Units live in metric names** (`_fahrenheit`, `_kwh`, `_gallons`,
`_dollars`), NOT via `metric.WithUnit` — the OTEL→Prometheus exporter would
otherwise append a second unit suffix (`..._fahrenheit_F`).
@@ -52,9 +70,9 @@ that's all it takes — no extra wiring.
- Regenerate proto: `make proto` (buf; `PROTO_DIRS=proto/econet/v1alpha1/*`).
- Regenerate config schema after config changes: `make schema`.
- Build/test: `go build ./... && go test ./...`.
- The rheemcloud `*Device` has no exported constructor, so `deviceToProto`
isn't unit-testable in isolation; test the pure helpers (`summarizeDevices`,
`b2i`) and exercise the rest with a live run.
- `econetclient.Device` is a plain struct, so `deviceToProto` and the datapoint
decoders (`mode`, `enabled`, `setpoint`, `hotWater`) are unit-testable by
hand-building a `Device` or feeding raw JSON to `newDevice`.
## Live verification (needs a real Rheem account)
+64 -15
View File
@@ -1,18 +1,67 @@
# v0.7.0
* feat: Add HTTP log exclusion regex paths to configuration schema.
* chore: Update Go version to 1.25.
* chore: Update module dependencies.
# Changelog
# v0.6.0
* feat: Introduce Model Context Protocol (MCP) server with a demo random fact tool.
* feat: Add MCP server configuration file (`contrib/mcpinspector.json`).
* deps: Update Go module dependencies, including `bufbuild/protovalidate`, `go-app`, `grpc-gateway`, `googleapis/api`, `grpc`, `protobuf`, `golang.org/x/exp`, and `googleapis/rpc`.
* deps: Add new Go module dependencies: `modelcontextprotocol/go-sdk` and `k8s.io/utils`.
All notable changes to econet-exporter are documented here. The format is
based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this
project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
# v0.5.0
## [v0.3.0] - 2026-07-05
* Added OpenTelemetry tracing for application startup.
* Updated Go module dependencies, including `protovalidate`, `go-app`, `grpc`, and OpenTelemetry related packages.
* Enhanced `Makefile` `rename` target to update the application name constant in `main.go`.
* Configured `air` live-reloading to exclude the `proto` directory.
* Refactored application initialization logic for improved modularity and OpenTelemetry integration.
### Changed
* Replaced the third-party `kevinburke/rheemcloud-go` dependency and its MQTT
coupling with an in-repo `econetclient` REST poller that talks directly to
the Rheem/EcoNet cloud REST API. Device state (including `@RUNNING`, `@MODE`,
and `@STATUS`) is sourced from the REST `getUserDataForApp` payload, so no
MQTT connection is required.
### Removed
* `rheemcloud` package and its `go.mod`/`go.sum` entries.
## [v0.2.0] - 2026-07-05
### Added
* `econetTLSInsecure` config option (env `ECONET_TLS_INSECURE`) to skip TLS
certificate verification when
connecting to the Rheem/EcoNet cloud API. The upstream host (cloudblade)
serves an old/untrusted certificate chain; this is a stopgap and should be
removed once the upstream certificate is fixed. See AGENTS.md.
* `TODO.md` tracking outstanding work (CA cert bundle, mutating proto RPCs,
additional tracing spans).
## [v0.1.2] - 2026-07-04
### Added
* CA certificate bundle baked into the container image so the exporter can
establish TLS connections to the EcoNet API.
### Changed
* Bumped Helm chart versions.
## [v0.1.1] - 2026-07-04
### Fixed
* Corrected the Helm chart `appVersion`.
## [v0.1.0] - 2026-07-04
Initial release of the EcoNet exporter for Rheem / Rheemcloud water heaters,
built on the `go-app` framework.
### Added
* `EconetService` owning a single shared `*rheemcloud.Client` across the gRPC,
MCP, and metrics sub-servers (fail-fast on missing credentials).
* gRPC API (`ListDevices`, `GetDevice`) with proto definitions and
grpc-gateway REST bindings.
* MCP server at `/api/mcp` exposing a `list_water_heaters` tool.
* OpenTelemetry observable gauges for device state (temperature, mode, energy
and water usage, cost) with a background usage poller, exported to
Prometheus. Units are encoded in metric names (`_fahrenheit`, `_kwh`,
`_gallons`, `_dollars`).
* `costPerKWH` config (US dollars per kWh) feeding `econet_energy_cost_dollars`.
* Helm chart for Kubernetes deployment.
+2
View File
@@ -19,6 +19,8 @@ FROM alpine:latest
ARG APP_NAME=econet-exporter
RUN apk add --no-cache ca-certificates
WORKDIR /app
USER 100:101
+15 -9
View File
@@ -1,8 +1,7 @@
# econet-exporter 🔥💧
An exporter for **Rheem EcoNet / Rheemcloud** water-heater data. It keeps a
long-lived connection to the Rheem cloud (via
[`rheemcloud-go`](https://github.com/kevinburke/rheemcloud-go)) and exposes device
An exporter for **Rheem EcoNet / Rheemcloud** water-heater data. It polls the
Rheem cloud REST API (ClearBlade backend) on an interval and exposes device
state and usage three ways:
- **OTEL metrics** served for Prometheus scraping at `/metrics` (live state +
@@ -25,7 +24,7 @@ go-app defaults. Custom fields (see `pkg/config/custom.go`):
| `econetEmail` | `ECONET_EMAIL` | Rheem account email |
| `econetPassword` | `ECONET_PASSWORD` | Prefer the env var; keep out of config.yaml |
| `costPerKWH` | `ECONET_COST_PER_KWH` | **US dollars per kWh** (e.g. `0.18` = 18¢), not cents |
| `usageInterval` | `ECONET_USAGE_INTERVAL` | Energy/water poll interval (default `5m`) |
| `pollInterval` | `ECONET_POLL_INTERVAL` | How often device state + daily energy/water usage are re-fetched over REST (default `1m`) |
Everything under the go-app `AppConfig` (logging, otel, http, grpc) is also
env-overridable with `APP_*` variables — see `env-sample`.
@@ -47,20 +46,27 @@ Then:
Per-device gauges labeled `econet.serial`, `econet.device_id`,
`econet.friendly_name`: `econet_setpoint_fahrenheit`, `econet_connected`,
`econet_running`, `econet_enabled`, `econet_away`,
`econet_hot_water_availability_percent`, `econet_alert_count`,
`econet_wifi_signal_db`, `econet_device_info` (adds `type`/`generic_type`/`mode`).
`econet_running` (adds a `running_state` label naming the active stage, e.g.
`compressor-running` / `element-running` / `idle`), `econet_enabled`,
`econet_away`, `econet_hot_water_availability_percent`, `econet_alert_count`,
`econet_wifi_signal_db`, `econet_mode` (enum: `1` for the current mode, `0`
otherwise, via a `mode` label), `econet_device_info` (adds
`type`/`generic_type`/`mode`).
Polled usage: `econet_energy_usage_kwh`, `econet_water_usage_gallons`,
`econet_energy_cost_dollars`.
> Note: `econet_wifi_signal_db` is only delivered over Rheem's MQTT push stream,
> which this exporter does not use — it stays at `0`.
## 📂 Project Structure
- `proto/econet/v1alpha1/` - Protobuf definitions (source).
- `api/econet/v1alpha1/` - Generated proto, grpc, grpc-gateway code.
- `pkg/config/` - Custom config merged with go-app configuration.
- `pkg/econet/` - `EconetService` (owns the rheemcloud client) plus:
- `pkg/econet/` - `EconetService` (owns the REST client + poll ticker) plus:
- `econetclient/` - minimal EcoNet REST client (auth, devices, usage).
- `econetgrpc/` - gRPC `EconetService` implementation.
- `econetmcp/` - MCP server + tools.
- `econetmetrics/` - OTEL observable gauges + usage poller.
- `econetmetrics/` - OTEL observable gauges.
- `helm/` - Helm chart for Kubernetes deployment.
- `.gitea/workflows/` - CI pipelines.
+6 -9
View File
@@ -1,10 +1,7 @@
# Demo app TODO
# econet-exporter TODO
- [ ] Fix Docerfile for ca cert bundle (TLS error to API host)
- [ ] Add mutating proto RPCs (change mode, etc..)
- [ ] Add spans to things
- [x] Create generic interface for implenting a service
- [x] Create config sample not called demo, so it is more easily reused
- [x] Update README for tagging/versioning/pipeline info
- [x] Update README for detail on installing protoc tools and make
- [x] Rename project
- [x] Finish grpc sample implementation
- [x] Add Dockerfile
- [x] Add gitea CI
+1 -1
View File
@@ -4,7 +4,7 @@
# NOTE: Prefer setting econetPassword via the ECONET_PASSWORD env var
econetEmail: ""
costPerKWH: 0.0 # US dollars per kWh (e.g. 0.18 = 18 cents), not cents
usageInterval: 5m
pollInterval: 1m # how often device state + daily usage are re-fetched over REST
# go-app config
name: econet-exporter
+6 -1
View File
@@ -124,6 +124,7 @@
},
"properties": {
"costPerKWH": {
"default": 0.19,
"type": "number"
},
"econetEmail": {
@@ -132,6 +133,10 @@
"econetPassword": {
"type": "string"
},
"econetTLSInsecure": {
"default": false,
"type": "boolean"
},
"environment": {
"type": "string"
},
@@ -150,7 +155,7 @@
"otel": {
"$ref": "#/definitions/ConfigOTELConfig"
},
"usageInterval": {
"pollInterval": {
"type": "integer"
},
"version": {
-5
View File
@@ -6,7 +6,6 @@ require (
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.11-20260415201107-50325440f8f2.1
gitea.libretechconsulting.com/rmcguire/go-app v0.17.1
github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0
github.com/kevinburke/rheemcloud-go v0.1.1
github.com/modelcontextprotocol/go-sdk v1.6.1
github.com/rs/zerolog v1.35.1
go.opentelemetry.io/otel/metric v1.44.0
@@ -43,14 +42,11 @@ require (
github.com/beorn7/perks v1.0.1 // indirect
github.com/cenkalti/backoff/v5 v5.0.3 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/eclipse/paho.mqtt.golang v1.5.1 // indirect
github.com/go-logr/logr v1.4.3 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/google/cel-go v0.29.1 // indirect
github.com/google/jsonschema-go v0.4.3 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/gorilla/websocket v1.5.3 // indirect
github.com/kevinburke/rest v0.0.0-20260604005914-d145d8a0294b // indirect
github.com/mattn/go-colorable v0.1.15 // indirect
github.com/mattn/go-isatty v0.0.22 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
@@ -70,7 +66,6 @@ require (
golang.org/x/exp v0.0.0-20260611194520-c48552f49976 // indirect
golang.org/x/net v0.56.0 // indirect
golang.org/x/oauth2 v0.36.0 // indirect
golang.org/x/sync v0.21.0 // indirect
golang.org/x/sys v0.46.0 // indirect
golang.org/x/text v0.38.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260630182238-925bb5da69e7 // indirect
-10
View File
@@ -24,8 +24,6 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UF
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE=
github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU=
github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeORc=
github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE=
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
@@ -45,18 +43,12 @@ github.com/google/jsonschema-go v0.4.3 h1:/DBOLZTfDow7pe2GmaJNhltueGTtDKICi8V8p+
github.com/google/jsonschema-go v0.4.3/go.mod h1:r5quNTdLOYEz95Ru18zA0ydNbBuYoo9tgaYcxEYhJVE=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3 h1:B+8ClL/kCQkRiU82d9xajRPKYMrB7E0MbtzWVi1K4ns=
github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.3/go.mod h1:NbCUVmiS4foBGBHOYlCT25+YmGpJ32dZPi75pGEUpj4=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs=
github.com/iancoleman/orderedmap v0.3.0 h1:5cbR2grmZR/DiVt+VJopEhtVs9YGInGIxAoMJn+Ichc=
github.com/iancoleman/orderedmap v0.3.0/go.mod h1:XuLcCUkdL5owUCQeF2Ue9uuw1EptkJDkXXS7VoV7XGE=
github.com/kevinburke/rest v0.0.0-20260604005914-d145d8a0294b h1:9ZWwhlz+lmPlXCz37dkRVMgK+iIwkFyQBogC2oWds/A=
github.com/kevinburke/rest v0.0.0-20260604005914-d145d8a0294b/go.mod h1:x35n2Fa5sM/IwpMDk6B9f7n6FkcBFX3yV9diIRfqxQc=
github.com/kevinburke/rheemcloud-go v0.1.1 h1:PsrpOso9rEeyATlAfDHjKx5E8C2TJseDwfeRvw4uzH4=
github.com/kevinburke/rheemcloud-go v0.1.1/go.mod h1:17QQqbT/YPQDnR3m+67EV7Wuve0onhKeiLAMcy1br1U=
github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
@@ -157,8 +149,6 @@ golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=
golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.38.0 h1:sXmwo9DwP3OK9EZ7PqAdaooSGozfl/3a6/xJcbzPRhE=
+2 -2
View File
@@ -15,13 +15,13 @@ type: application
# This is the chart version. This version number should be incremented each time you make changes
# to the chart and its templates, including the app version.
# Versions are expected to follow Semantic Versioning (https://semver.org/)
version: 0.1.3
version: 0.1.6
# This is the version number of the application being deployed. This version number should be
# incremented each time you make changes to the application. Versions are not expected to
# follow Semantic Versioning. They should reflect the version the application is using.
# It is recommended to use it with quotes.
appVersion: "v0.7.0"
appVersion: "v0.3.0"
dependencies:
- name: hull
+1 -1
View File
@@ -7,7 +7,7 @@ hull:
# NOTE: Prefer supplying econetPassword via the ECONET_PASSWORD env var
econetEmail: ""
costPerKWH: 0.19
usageInterval: 5m
pollInterval: 1m
# go-app config
name: econet-exporter
logging:
+14 -10
View File
@@ -12,9 +12,9 @@ import (
"gitea.libretechconsulting.com/rmcguire/go-app/pkg/config"
)
// DefaultUsageInterval is used when UsageInterval is unset. Rheem's
// energy/water history only updates hourly, so polling faster is wasteful.
const DefaultUsageInterval = 5 * time.Minute
// DefaultPollInterval is used when PollInterval is unset. Each tick re-fetches
// device state and daily energy/water usage over REST.
const DefaultPollInterval = 1 * time.Minute
type ServiceConfig struct {
// Rheem EcoNet cloud credentials. Prefer supplying the password via
@@ -22,13 +22,17 @@ type ServiceConfig struct {
EconetEmail string `yaml:"econetEmail" json:"econetEmail,omitempty" env:"ECONET_EMAIL"`
EconetPassword string `yaml:"econetPassword" json:"econetPassword,omitempty" env:"ECONET_PASSWORD"`
// Necessary due to ancient TLS certificate run by cloudblade that isn't trusted in modern ca-certificate bundles
EconetTLSInsecure bool `yaml:"econetTLSInsecure" json:"econetTLSInsecure,omitempty" env:"ECONET_TLS_INSECURE" default:"false"`
// CostPerKWH is the electricity price in US dollars per kWh (e.g. 0.18
// means 18 cents/kWh, NOT 18). It derives the econet_energy_cost_dollars
// metric from kWh energy usage.
CostPerKWH float64 `yaml:"costPerKWH" json:"costPerKWH,omitempty" env:"ECONET_COST_PER_KWH" default:"0.19"`
// UsageInterval controls how often energy/water usage history is polled.
UsageInterval time.Duration `yaml:"usageInterval" json:"usageInterval,omitempty" env:"ECONET_USAGE_INTERVAL"`
// PollInterval controls how often device state and daily energy/water
// usage are re-fetched over REST.
PollInterval time.Duration `yaml:"pollInterval" json:"pollInterval,omitempty" env:"ECONET_POLL_INTERVAL"`
// Embeds go-app config, used by go-app to
// merge custom config into go-app config, and to produce
@@ -47,10 +51,10 @@ func (c *ServiceConfig) LoadEnv() error {
return env.Parse(c)
}
// GetUsageInterval returns the configured usage poll interval or the default.
func (c *ServiceConfig) GetUsageInterval() time.Duration {
if c.UsageInterval <= 0 {
return DefaultUsageInterval
// GetPollInterval returns the configured poll interval or the default.
func (c *ServiceConfig) GetPollInterval() time.Duration {
if c.PollInterval <= 0 {
return DefaultPollInterval
}
return c.UsageInterval
return c.PollInterval
}
+43 -22
View File
@@ -6,14 +6,15 @@ package econet
import (
"context"
"fmt"
"log/slog"
"time"
rheemcloud "github.com/kevinburke/rheemcloud-go"
"github.com/rs/zerolog"
optsgrpc "gitea.libretechconsulting.com/rmcguire/go-app/pkg/srv/grpc/opts"
optshttp "gitea.libretechconsulting.com/rmcguire/go-app/pkg/srv/http/opts"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/config"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/econet/econetclient"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/econet/econetgrpc"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/econet/econetmcp"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/econet/econetmetrics"
@@ -23,7 +24,8 @@ import (
type EconetService struct {
ctx context.Context
config *config.ServiceConfig
client *rheemcloud.Client
log *zerolog.Logger
client *econetclient.Client
grpc *econetgrpc.EconetGRPCServer
mcp *econetmcp.EconetMCPServer
metrics *econetmetrics.Collector
@@ -32,39 +34,58 @@ type EconetService struct {
func (e *EconetService) Init(ctx context.Context, cfg *config.ServiceConfig) (service.ShutdownFunc, error) {
e.ctx = ctx
e.config = cfg
e.log = zerolog.Ctx(ctx)
client, err := connect(ctx, cfg)
if err != nil {
return nil, err
// Fail fast on missing credentials (cheap, no network).
if cfg.EconetEmail == "" || cfg.EconetPassword == "" {
return nil, fmt.Errorf("econet: email and password required (set ECONET_EMAIL / ECONET_PASSWORD)")
}
e.client = client
e.grpc = econetgrpc.NewEconetGRPCServer(ctx, cfg, client)
e.client = econetclient.New(cfg, e.log)
e.grpc = econetgrpc.NewEconetGRPCServer(ctx, cfg, e.client)
e.mcp = econetmcp.NewEconetMCPServer(ctx, cfg, e.grpc)
e.metrics = econetmetrics.NewCollector(ctx, cfg, client)
if err := e.metrics.Start(); err != nil {
e.metrics = econetmetrics.NewCollector(ctx, cfg, e.client)
if err := e.metrics.RegisterGauges(); err != nil {
return nil, err
}
// Poll the cloud in the background so a slow or unreachable Rheem cloud
// never blocks HTTP/gRPC server startup. Devices are empty until the first
// successful refresh; the gauges and gRPC handlers tolerate that.
go e.pollLoop(ctx)
return e.shutdown, nil
}
func connect(ctx context.Context, cfg *config.ServiceConfig) (*rheemcloud.Client, error) {
if cfg.EconetEmail == "" || cfg.EconetPassword == "" {
return nil, fmt.Errorf("econet: email and password required (set ECONET_EMAIL / ECONET_PASSWORD)")
// pollLoop refreshes device state on a ticker for the life of the service.
// A per-refresh timeout keeps a hung REST call from stalling the loop.
func (e *EconetService) pollLoop(ctx context.Context) {
interval := e.config.GetPollInterval()
e.refresh(ctx)
t := time.NewTicker(interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
e.refresh(ctx)
}
}
client, err := rheemcloud.Connect(ctx, cfg.EconetEmail, cfg.EconetPassword, &rheemcloud.Config{
Logger: slog.Default(),
})
if err != nil {
return nil, fmt.Errorf("econet: connect failed: %w", err)
}
func (e *EconetService) refresh(ctx context.Context) {
rctx, cancel := context.WithTimeout(ctx, e.config.GetPollInterval())
defer cancel()
if err := e.client.Refresh(rctx); err != nil {
e.log.Error().Err(err).Msg("econet: refresh failed")
return
}
return client, nil
e.log.Debug().Int("devices", len(e.client.Devices())).Msg("econet: refreshed")
}
func (e *EconetService) shutdown(_ context.Context) (string, error) {
e.metrics.Stop()
return "EconetService", e.client.Close()
return "EconetService", nil
}
func (e *EconetService) GetGRPC() *optsgrpc.AppGRPC {
@@ -86,7 +107,7 @@ func (e *EconetService) healthChecks() []optshttp.HealthCheckFunc {
return []optshttp.HealthCheckFunc{
func(_ context.Context) error {
if len(e.client.Devices()) == 0 {
return fmt.Errorf("econet: no devices loaded")
return fmt.Errorf("econet: no devices loaded yet")
}
return nil
},
+324
View File
@@ -0,0 +1,324 @@
// Package econetclient is a minimal REST client for the Rheem EcoNet cloud
// (ClearBlade backend). It authenticates, lists devices, and reads daily
// energy/water usage over plain HTTPS — no MQTT. Callers drive it by calling
// Refresh on a ticker and reading the decoded Device snapshots.
package econetclient
import (
"bytes"
"context"
"crypto/tls"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"sync"
"time"
"github.com/rs/zerolog"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/config"
)
// Static credentials embedded in the Rheem EcoNet Android app; the ClearBlade
// backend uses them to identify the calling app independent of the per-user
// auth. Same values pyeconet and kevinburke/rheemcloud-go use.
const (
clearBladeSystemKey = "e2e699cb0bb0bbb88fc8858cb5a401"
clearBladeSystemSecret = "E2E699CB0BE6C6FADDB1B0BC9A20"
baseURL = "https://rheem.clearblade.com/api/v/1"
)
// Client talks to the EcoNet REST API and caches the latest device snapshot.
// It is safe for concurrent use: Refresh swaps the device map under a lock,
// and readers take a read lock.
type Client struct {
email string
password string
costPerKWH float64
http *http.Client
log *zerolog.Logger
mu sync.RWMutex
token string
accountID string
devices map[string]*Device
}
// New builds a Client from config. It performs no network I/O; call Refresh
// to authenticate and populate devices.
func New(cfg *config.ServiceConfig, log *zerolog.Logger) *Client {
transport := http.DefaultTransport.(*http.Transport).Clone()
transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: cfg.EconetTLSInsecure} //nolint:gosec // opt-in via EconetTLSInsecure
return &Client{
email: cfg.EconetEmail,
password: cfg.EconetPassword,
costPerKWH: cfg.CostPerKWH,
http: &http.Client{Transport: transport, Timeout: 15 * time.Second},
log: log,
devices: map[string]*Device{},
}
}
// CostPerKWH exposes the configured electricity price for the usage gauges.
func (c *Client) CostPerKWH() float64 { return c.costPerKWH }
// Devices returns a snapshot slice of the currently-known devices. Empty until
// the first successful Refresh.
func (c *Client) Devices() []*Device {
c.mu.RLock()
defer c.mu.RUnlock()
out := make([]*Device, 0, len(c.devices))
for _, d := range c.devices {
out = append(out, d)
}
return out
}
// Device looks up a single device by serial number, or nil if unknown.
func (c *Client) Device(serial string) *Device {
c.mu.RLock()
defer c.mu.RUnlock()
return c.devices[serial]
}
// Refresh authenticates if needed, re-fetches the device list, reads each
// unit's daily usage, and atomically replaces the cached snapshot. On an auth
// failure it clears the token so the next call re-authenticates.
func (c *Client) Refresh(ctx context.Context) error {
if err := c.ensureAuth(ctx); err != nil {
return err
}
equipment, err := c.fetchDevices(ctx)
if err != nil {
c.clearToken() // force re-auth next time in case the token expired
return err
}
now := time.Now()
devices := make(map[string]*Device, len(equipment))
for _, raw := range equipment {
d := newDevice(raw, now)
if d.SerialNumber == "" {
continue
}
c.fetchUsage(ctx, d)
devices[d.SerialNumber] = d
}
c.mu.Lock()
c.devices = devices
c.mu.Unlock()
return nil
}
func (c *Client) clearToken() {
c.mu.Lock()
c.token = ""
c.mu.Unlock()
}
func (c *Client) ensureAuth(ctx context.Context) error {
c.mu.RLock()
have := c.token != ""
c.mu.RUnlock()
if have {
return nil
}
return c.authenticate(ctx)
}
// --- REST calls ----------------------------------------------------------
func (c *Client) authenticate(ctx context.Context) error {
body := map[string]string{"email": c.email, "password": c.password}
var ar struct {
UserToken string `json:"user_token"`
Options struct {
Success bool `json:"success"`
AccountID string `json:"account_id"`
Message string `json:"message"`
} `json:"options"`
}
if err := c.do(ctx, "/user/auth", "", body, &ar); err != nil {
return fmt.Errorf("econet: auth: %w", err)
}
if !ar.Options.Success || ar.UserToken == "" || ar.Options.AccountID == "" {
if ar.Options.Message != "" {
return fmt.Errorf("econet: auth rejected: %s", ar.Options.Message)
}
return fmt.Errorf("econet: auth rejected (empty token/account)")
}
c.mu.Lock()
c.token = ar.UserToken
c.accountID = ar.Options.AccountID
c.mu.Unlock()
return nil
}
// fetchDevices returns the raw equipment blocks from getUserDataForApp.
func (c *Client) fetchDevices(ctx context.Context) ([]map[string]json.RawMessage, error) {
var resp struct {
Results struct {
Locations []struct {
// "equiptments" is misspelled upstream; match it exactly.
Equipments []map[string]json.RawMessage `json:"equiptments"`
} `json:"locations"`
} `json:"results"`
}
path := "/code/" + clearBladeSystemKey + "/getUserDataForApp"
if err := c.do(ctx, path, c.userToken(), map[string]string{"resource": "friedrich"}, &resp); err != nil {
return nil, fmt.Errorf("econet: device list: %w", err)
}
var out []map[string]json.RawMessage
for _, loc := range resp.Results.Locations {
out = append(out, loc.Equipments...)
}
return out, nil
}
// fetchUsage reads the device's daily energy and water totals, filling the
// usage fields. Failures are logged at debug and leave the fields at zero
// (e.g. units that don't track water return an error).
func (c *Client) fetchUsage(ctx context.Context, d *Device) {
if e, typ, err := c.usage(ctx, d, "energyUsage"); err == nil {
d.EnergyKWH, d.EnergyType = e, typ
} else {
c.log.Debug().Err(err).Str("serial", d.SerialNumber).Msg("energy usage poll failed")
}
if w, _, err := c.usage(ctx, d, "waterUsage"); err == nil {
d.WaterGallons = w
} else {
c.log.Debug().Err(err).Str("serial", d.SerialNumber).Msg("water usage poll failed")
}
}
// usage POSTs a dynamicAction usage report and sums today's buckets. The
// energy unit ("KWH"/"KBTU") is inferred from the summary message.
func (c *Client) usage(ctx context.Context, d *Device, usageType string) (total float64, energyType string, err error) {
start, end := dayWindow(usageType == "energyUsage")
payload := map[string]string{
"ACTION": "waterheaterUsageReportView",
"device_name": d.DeviceID,
"serial_number": d.SerialNumber,
"start_date": start,
"end_date": end,
"usage_type": usageType,
}
var raw json.RawMessage
path := "/code/" + clearBladeSystemKey + "/dynamicAction"
if err := c.do(ctx, path, c.userToken(), payload, &raw); err != nil {
return 0, "", err
}
var env struct {
Success bool `json:"success"`
Error string `json:"error"`
Results struct {
Energy usageBlock `json:"energy_usage"`
Water usageBlock `json:"water_usage"`
} `json:"results"`
}
if err := json.Unmarshal(raw, &env); err != nil {
return 0, "", fmt.Errorf("parse usage: %w", err)
}
if !env.Success {
if env.Error != "" {
return 0, "", fmt.Errorf("usage rejected: %s", env.Error)
}
return 0, "", fmt.Errorf("usage rejected")
}
block := env.Results.Water
if usageType == "energyUsage" {
block = env.Results.Energy
}
for _, p := range block.Data {
total += p.Value
}
return total, energyTypeFromMessage(block.Message, d.GenericType), nil
}
type usageBlock struct {
Data []struct {
Value float64 `json:"value"`
} `json:"data"`
Message string `json:"message"`
}
// do POSTs a JSON body to path and decodes the JSON response into out. When
// token is non-empty it is sent as the ClearBlade-UserToken header.
func (c *Client) do(ctx context.Context, path, token string, body, out any) error {
b, err := json.Marshal(body)
if err != nil {
return err
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+path, bytes.NewReader(b))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("ClearBlade-SystemKey", clearBladeSystemKey)
req.Header.Set("ClearBlade-SystemSecret", clearBladeSystemSecret)
if token != "" {
req.Header.Set("ClearBlade-UserToken", token)
}
resp, err := c.http.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
raw, err := io.ReadAll(resp.Body)
if err != nil {
return err
}
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("HTTP %d: %s", resp.StatusCode, bytes.TrimSpace(raw))
}
return json.Unmarshal(raw, out)
}
func (c *Client) userToken() string {
c.mu.RLock()
defer c.mu.RUnlock()
return c.token
}
// --- usage helpers -------------------------------------------------------
// dayWindow returns ISO-8601 start/end strings covering the current local day,
// matching pyeconet's formats: energy includes the timezone offset, water does
// not (the upstream parser is picky).
func dayWindow(energy bool) (start, end string) {
now := time.Now()
s := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 999_000_000, now.Location())
e := time.Date(now.Year(), now.Month(), now.Day(), 23, 59, 59, 999_000_000, now.Location())
if energy {
return pyISOMillis(s), pyISOMillis(e)
}
const f = "2006-01-02T15:04:05.999"
return s.Format(f), e.Format(f)
}
// pyISOMillis emulates Python's datetime.isoformat(timespec='milliseconds')
// for a tz-aware datetime: YYYY-MM-DDTHH:MM:SS.mmm±HH:MM.
func pyISOMillis(t time.Time) string {
s := t.Format("2006-01-02T15:04:05.000-0700")
if len(s) >= 5 { // "-0700" -> "-07:00"
s = s[:len(s)-2] + ":" + s[len(s)-2:]
}
return s
}
// energyTypeFromMessage extracts the unit label ("KWH"/"KBTU") from Rheem's
// human summary ("You used X.Y kWh of energy today."), the 4th word per
// pyeconet's heuristic, falling back to the device's generic type.
func energyTypeFromMessage(msg, genericType string) string {
fields := strings.Fields(msg)
if len(fields) >= 4 {
return strings.ToUpper(fields[3])
}
if genericType == "gasWaterHeater" {
return "KBTU"
}
return "KWH"
}
+322
View File
@@ -0,0 +1,322 @@
package econetclient
import (
"encoding/json"
"strconv"
"strings"
"time"
)
// Device is a flat snapshot of one piece of Rheem equipment, decoded from
// the getUserDataForApp REST response. Fields are computed once per refresh;
// consumers read them directly. WiFiSignal only arrives over MQTT (which this
// client does not use), so it stays at its zero value.
type Device struct {
SerialNumber string
DeviceID string
FriendlyName string
Type string // water_heater | thermostat | unknown
GenericType string // e.g. heatpumpWaterHeater, gasWaterHeater
Connected bool
WiFiSignal int // MQTT-only; always 0 here
Mode string
Enabled bool
Running bool // true when actively heating (decoded from @RUNNING over REST)
RunningState string // normalized @RUNNING label: "compressor-running" | "element-running" | "idle" | ...
Setpoint int
SetpointMin int
SetpointMax int
HotWaterAvailability int // -1 if not reported, else 0/33/66/100
AlertCount int
Away bool
LastUpdated time.Time
// Usage totals for the current day, populated by Client.Refresh.
EnergyKWH float64
EnergyType string // "KWH" or "KBTU"
WaterGallons float64
}
// newDevice decodes an equipment block (the raw JSON map for one unit) into
// a flat Device. It ports the accessor logic from kevinburke/rheemcloud-go
// so behaviour matches, minus the MQTT-only fields.
func newDevice(info map[string]json.RawMessage, now time.Time) *Device {
d := &Device{
DeviceID: jsonStr(info["device_name"]),
SerialNumber: jsonStr(info["serial_number"]),
Type: deviceTypeString(jsonStr(info["device_type"])),
GenericType: jsonStr(info["@TYPE"]),
Connected: connected(info),
Away: jsonBool(info["@AWAY"]),
AlertCount: alertCount(info["@ALERTCOUNT"]),
HotWaterAvailability: hotWater(info["@HOTWATER"]),
Enabled: enabled(info),
Mode: mode(info),
LastUpdated: now,
}
d.FriendlyName = friendlyName(info, d.DeviceID)
d.Setpoint, d.SetpointMin, d.SetpointMax = setpoint(info["@SETPOINT"])
d.Running, d.RunningState = running(info)
return d
}
// --- field decoders ------------------------------------------------------
func connected(info map[string]json.RawMessage) bool {
v, ok := info["@CONNECTED"]
if !ok {
return true // older units omit it and are assumed connected
}
return jsonBool(v)
}
func friendlyName(info map[string]json.RawMessage, deviceID string) string {
if obj := decodeDatapoint(info["@NAME"]); obj != nil {
if s := jsonStr(obj.Value); s != "" {
return s
}
}
return deviceID
}
func alertCount(v json.RawMessage) int {
if len(v) == 0 {
return 0
}
var n float64
if json.Unmarshal(v, &n) == nil {
return int(n)
}
var s string
if json.Unmarshal(v, &s) == nil {
if i, err := strconv.Atoi(s); err == nil {
return i
}
}
return 0
}
func hotWater(v json.RawMessage) int {
var icon string
if len(v) == 0 || json.Unmarshal(v, &icon) != nil {
return -1
}
switch {
case strings.Contains(icon, "ic_tank_hundread_percent"):
return 100
case strings.Contains(icon, "ic_tank_fourty_percent"):
return 66
case strings.Contains(icon, "ic_tank_ten_percent"):
return 33
case strings.Contains(icon, "ic_tank_empty"), strings.Contains(icon, "ic_tank_zero_percent"):
return 0
default:
return -1
}
}
func setpoint(v json.RawMessage) (val, min, max int) {
obj := decodeDatapoint(v)
if obj == nil {
return 0, 0, 0
}
var n float64
_ = json.Unmarshal(obj.Value, &n)
return int(n), int(obj.Constraints.LowerLimit), int(obj.Constraints.UpperLimit)
}
// enabled reports the master on/off state, porting rheemcloud's Device.Enabled.
func enabled(info map[string]json.RawMessage) bool {
if _, ok := info["@MODE"]; ok {
labels := modeLabels(info)
if obj := decodeDatapoint(info["@MODE"]); obj != nil {
if idx, ok := datapointIndex(obj); ok && idx >= 0 && idx < len(labels) {
return !strings.EqualFold(labels[idx], "OFF")
}
}
}
if _, ok := info["@ENABLED"]; ok {
obj := decodeDatapoint(info["@ENABLED"])
if obj == nil {
return false
}
idx, _ := datapointIndex(obj)
return idx == 1
}
return false
}
// mode returns the current operating-mode label, porting rheemcloud's Device.Mode.
func mode(info map[string]json.RawMessage) string {
_, supportsModes := info["@MODE"]
_, supportsOnOff := info["@ENABLED"]
if supportsOnOff && !enabled(info) {
return "off"
}
if !supportsModes {
switch jsonStr(info["@TYPE"]) {
case "gasWaterHeater", "tanklessWaterHeater":
return "gas"
default:
return "electric"
}
}
labels := modeLabels(info)
obj := decodeDatapoint(info["@MODE"])
if obj == nil {
return "unknown"
}
idx, _ := datapointIndex(obj)
if idx < 0 || idx >= len(labels) {
return "unknown"
}
return modeFromEnumText(labels[idx])
}
func modeLabels(info map[string]json.RawMessage) []string {
obj := decodeDatapoint(info["@MODE"])
if obj == nil {
return nil
}
return obj.Constraints.EnumText
}
// modeFromEnumText normalizes a Rheem mode label into a stable slug,
// matching pyeconet's WaterHeaterOperationMode labels.
func modeFromEnumText(s string) string {
cleaned := strings.ToUpper(strings.ReplaceAll(strings.ReplaceAll(strings.TrimRight(s, " "), " ", "_"), "/", "_"))
switch cleaned {
case "OFF":
return "off"
case "ELECTRIC", "ELECTRIC_MODE":
return "electric"
case "ENERGY_SAVING", "ENERGY_SAVER":
return "energy-saving"
case "HEAT_PUMP", "HEAT_PUMP_ONLY":
return "heat-pump"
case "HIGH_DEMAND":
return "high-demand"
case "GAS":
return "gas"
case "PERFORMANCE":
return "performance"
case "VACATION":
return "vacation"
default:
return "unknown"
}
}
// KnownModes are the normalized operating-mode slugs modeFromEnumText emits.
// Consumers (e.g. the mode enum gauge) iterate these so every mode gets a
// series regardless of which one is currently active.
var KnownModes = []string{
"off", "electric", "energy-saving", "heat-pump",
"high-demand", "gas", "performance", "vacation", "unknown",
}
// running reports whether the unit is actively heating and a normalized label
// for what is running. It reads the @RUNNING datapoint, a bare string the REST
// getUserDataForApp payload populates (e.g. "Compressor Running" -> true,
// "compressor-running"); empty means idle. This is served over plain REST
// despite older code assuming it was MQTT-only.
func running(info map[string]json.RawMessage) (active bool, state string) {
raw := strings.TrimSpace(jsonStr(info["@RUNNING"]))
if raw == "" {
return false, "idle"
}
state = slugify(raw)
if state == "off" || state == "idle" {
return false, state
}
return true, state
}
// slugify normalizes a Rheem label into a stable lowercase kebab-case slug,
// e.g. "Compressor Running" -> "compressor-running".
func slugify(s string) string {
s = strings.ToLower(strings.TrimSpace(s))
s = strings.ReplaceAll(s, "/", "-")
s = strings.Join(strings.Fields(s), "-")
return s
}
func deviceTypeString(s string) string {
switch s {
case "WH":
return "water_heater"
case "HVAC":
return "thermostat"
default:
return "unknown"
}
}
// --- datapoint plumbing --------------------------------------------------
// datapointObject is the {value, status, title, constraints} shape many
// "@..." fields use. Some fields are bare scalars instead; decodeDatapoint
// returns nil for those and callers fall back to reading the raw value.
type datapointObject struct {
Value json.RawMessage `json:"value"`
Status string `json:"status"`
Title string `json:"title"`
Constraints datapointConstraints `json:"constraints"`
}
type datapointConstraints struct {
LowerLimit float64 `json:"lowerLimit"`
UpperLimit float64 `json:"upperLimit"`
EnumText []string `json:"enumText"`
}
func decodeDatapoint(b json.RawMessage) *datapointObject {
if len(b) == 0 || !isJSONObject(b) {
return nil
}
var obj datapointObject
if json.Unmarshal(b, &obj) != nil {
return nil
}
return &obj
}
// datapointIndex reads a datapoint's numeric value (mode index / enabled flag).
func datapointIndex(obj *datapointObject) (int, bool) {
var n float64
if json.Unmarshal(obj.Value, &n) != nil {
return 0, false
}
return int(n), true
}
func isJSONObject(b json.RawMessage) bool {
for _, c := range b {
switch c {
case ' ', '\t', '\n', '\r':
continue
case '{':
return true
default:
return false
}
}
return false
}
func jsonStr(v json.RawMessage) string {
var s string
_ = json.Unmarshal(v, &s)
return s
}
func jsonBool(v json.RawMessage) bool {
var b bool
_ = json.Unmarshal(v, &b)
return b
}
+19 -20
View File
@@ -1,32 +1,31 @@
package econetgrpc
import (
rheemcloud "github.com/kevinburke/rheemcloud-go"
"google.golang.org/protobuf/types/known/timestamppb"
pb "gitea.libretechconsulting.com/rmcguire/econet-exporter/api/econet/v1alpha1"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/econet/econetclient"
)
func deviceToProto(d *rheemcloud.Device) *pb.Device {
min, max := d.SetpointLimits()
func deviceToProto(d *econetclient.Device) *pb.Device {
return &pb.Device{
SerialNumber: d.SerialNumber(),
DeviceId: d.DeviceID(),
FriendlyName: d.FriendlyName(),
Type: d.Type().String(),
GenericType: d.GenericType(),
Connected: d.Connected(),
WifiSignal: int32(d.WiFiSignal()),
Mode: d.Mode().String(),
Enabled: d.Enabled(),
Running: d.Running(),
RunningState: d.RunningState(),
Setpoint: int32(d.Setpoint()),
SetpointMin: int32(min),
SetpointMax: int32(max),
HotWaterAvailability: int32(d.HotWaterAvailability()),
AlertCount: int32(d.AlertCount()),
Away: d.Away(),
SerialNumber: d.SerialNumber,
DeviceId: d.DeviceID,
FriendlyName: d.FriendlyName,
Type: d.Type,
GenericType: d.GenericType,
Connected: d.Connected,
WifiSignal: int32(d.WiFiSignal),
Mode: d.Mode,
Enabled: d.Enabled,
Running: d.Running,
RunningState: d.RunningState,
Setpoint: int32(d.Setpoint),
SetpointMin: int32(d.SetpointMin),
SetpointMax: int32(d.SetpointMax),
HotWaterAvailability: int32(d.HotWaterAvailability),
AlertCount: int32(d.AlertCount),
Away: d.Away,
LastUpdated: timestamppb.New(d.LastUpdated),
}
}
+4 -4
View File
@@ -1,11 +1,10 @@
// Package econetgrpc implements the EconetService gRPC API over a shared
// rheemcloud.Client, exposing read-only device state.
// econetclient.Client, exposing read-only device state.
package econetgrpc
import (
"context"
rheemcloud "github.com/kevinburke/rheemcloud-go"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
@@ -16,17 +15,18 @@ import (
pb "gitea.libretechconsulting.com/rmcguire/econet-exporter/api/econet/v1alpha1"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/config"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/econet/econetclient"
)
type EconetGRPCServer struct {
tracer trace.Tracer
ctx context.Context
cfg *config.ServiceConfig
client *rheemcloud.Client
client *econetclient.Client
pb.UnimplementedEconetServiceServer
}
func NewEconetGRPCServer(ctx context.Context, cfg *config.ServiceConfig, client *rheemcloud.Client) *EconetGRPCServer {
func NewEconetGRPCServer(ctx context.Context, cfg *config.ServiceConfig, client *econetclient.Client) *EconetGRPCServer {
return &EconetGRPCServer{
ctx: ctx,
cfg: cfg,
+42 -39
View File
@@ -3,9 +3,10 @@ package econetmetrics
import (
"context"
rheemcloud "github.com/kevinburke/rheemcloud-go"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/econet/econetclient"
)
// registerLive wires the per-device live-state gauges to a single callback
@@ -17,13 +18,14 @@ func (c *Collector) registerLive() error {
setpointMin: b.i64("econet_setpoint_min_fahrenheit", "Minimum allowed setpoint (°F)"),
setpointMax: b.i64("econet_setpoint_max_fahrenheit", "Maximum allowed setpoint (°F)"),
connected: b.i64("econet_connected", "1 if the device is connected"),
running: b.i64("econet_running", "1 if the device is actively heating"),
running: b.i64("econet_running", "1 if the device is actively heating (running_state label names the stage, e.g. compressor-running)"),
mode: b.i64("econet_mode", "Operating mode enum: 1 for the current mode, 0 otherwise (see the mode label)"),
enabled: b.i64("econet_enabled", "1 if the device is enabled"),
away: b.i64("econet_away", "1 if away mode is on"),
hotWater: b.i64("econet_hot_water_availability_percent", "Hot water availability (%)"),
alerts: b.i64("econet_alert_count", "Number of active alerts"),
wifi: b.i64("econet_wifi_signal_db", "WiFi signal strength (dB)"),
lastUpdated: b.i64("econet_last_updated_seconds", "Unix time of last MQTT state update"),
lastUpdated: b.i64("econet_last_updated_seconds", "Unix time of last state update"),
info: b.i64("econet_device_info", "Device metadata (value is always 1)"),
}
if b.err != nil {
@@ -38,8 +40,8 @@ func (c *Collector) registerLive() error {
return err
}
// registerUsage wires the polled energy/water gauges, reading the cache the
// poller refreshes (the underlying REST calls are too slow for a callback).
// registerUsage wires the energy/water gauges, reading the usage totals the
// poller folds into each device snapshot.
func (c *Collector) registerUsage() error {
b := &gaugeBuilder{m: c.meter}
g := &usageGauges{
@@ -51,10 +53,8 @@ func (c *Collector) registerUsage() error {
return b.err
}
_, err := c.meter.RegisterCallback(func(_ context.Context, o metric.Observer) error {
c.mu.RLock()
defer c.mu.RUnlock()
for serial, u := range c.usage {
g.observe(o, serial, u, c.cfg.CostPerKWH)
for _, d := range c.client.Devices() {
g.observe(o, d, c.client.CostPerKWH())
}
return nil
}, b.obs...)
@@ -65,27 +65,30 @@ type liveGauges struct {
setpoint, setpointMin, setpointMax metric.Int64ObservableGauge
connected, running, enabled, away metric.Int64ObservableGauge
hotWater, alerts, wifi, lastUpdated metric.Int64ObservableGauge
mode metric.Int64ObservableGauge
info metric.Int64ObservableGauge
}
func (g *liveGauges) observe(o metric.Observer, d *rheemcloud.Device) {
func (g *liveGauges) observe(o metric.Observer, d *econetclient.Device) {
set := metric.WithAttributes(deviceAttrs(d)...)
min, max := d.SetpointLimits()
o.ObserveInt64(g.setpoint, int64(d.Setpoint()), set)
o.ObserveInt64(g.setpointMin, int64(min), set)
o.ObserveInt64(g.setpointMax, int64(max), set)
o.ObserveInt64(g.connected, b2i(d.Connected()), set)
o.ObserveInt64(g.running, b2i(d.Running()), set)
o.ObserveInt64(g.enabled, b2i(d.Enabled()), set)
o.ObserveInt64(g.away, b2i(d.Away()), set)
o.ObserveInt64(g.hotWater, int64(d.HotWaterAvailability()), set)
o.ObserveInt64(g.alerts, int64(d.AlertCount()), set)
o.ObserveInt64(g.wifi, int64(d.WiFiSignal()), set)
// LastUpdated is zero until the first MQTT update lands; skip it then
// rather than emit a nonsensical negative epoch.
o.ObserveInt64(g.setpoint, int64(d.Setpoint), set)
o.ObserveInt64(g.setpointMin, int64(d.SetpointMin), set)
o.ObserveInt64(g.setpointMax, int64(d.SetpointMax), set)
o.ObserveInt64(g.connected, b2i(d.Connected), set)
runSet := metric.WithAttributes(append(deviceAttrs(d), attribute.String("econet.running_state", d.RunningState))...)
o.ObserveInt64(g.running, b2i(d.Running), runSet)
o.ObserveInt64(g.enabled, b2i(d.Enabled), set)
o.ObserveInt64(g.away, b2i(d.Away), set)
o.ObserveInt64(g.hotWater, int64(d.HotWaterAvailability), set)
o.ObserveInt64(g.alerts, int64(d.AlertCount), set)
o.ObserveInt64(g.wifi, int64(d.WiFiSignal), set)
if !d.LastUpdated.IsZero() {
o.ObserveInt64(g.lastUpdated, d.LastUpdated.Unix(), set)
}
for _, m := range econetclient.KnownModes {
modeSet := metric.WithAttributes(append(deviceAttrs(d), attribute.String("econet.mode", m))...)
o.ObserveInt64(g.mode, b2i(m == d.Mode), modeSet)
}
o.ObserveInt64(g.info, 1, metric.WithAttributes(infoAttrs(d)...))
}
@@ -93,34 +96,34 @@ type usageGauges struct {
energy, cost, water metric.Float64ObservableGauge
}
func (g *usageGauges) observe(o metric.Observer, serial string, u usage, costPerKWH float64) {
set := metric.WithAttributes(attribute.String("econet.serial", serial))
o.ObserveFloat64(g.energy, u.energyKWH, metric.WithAttributes(
attribute.String("econet.serial", serial),
attribute.String("econet.energy_type", u.energyType),
func (g *usageGauges) observe(o metric.Observer, d *econetclient.Device, costPerKWH float64) {
serial := attribute.String("econet.serial", d.SerialNumber)
o.ObserveFloat64(g.energy, d.EnergyKWH, metric.WithAttributes(
serial,
attribute.String("econet.energy_type", d.EnergyType),
))
o.ObserveFloat64(g.water, u.waterGallons, set)
if u.energyType == "KWH" {
o.ObserveFloat64(g.cost, u.energyKWH*costPerKWH, set)
o.ObserveFloat64(g.water, d.WaterGallons, metric.WithAttributes(serial))
if d.EnergyType == "KWH" {
o.ObserveFloat64(g.cost, d.EnergyKWH*costPerKWH, metric.WithAttributes(serial))
}
}
// deviceAttrs are low-cardinality identity attributes safe for every series.
// They follow OTEL semantic-convention style (namespaced, dotted keys) and
// deliberately exclude anything that changes over time (e.g. timestamps).
func deviceAttrs(d *rheemcloud.Device) []attribute.KeyValue {
func deviceAttrs(d *econetclient.Device) []attribute.KeyValue {
return []attribute.KeyValue{
attribute.String("econet.serial", d.SerialNumber()),
attribute.String("econet.device_id", d.DeviceID()),
attribute.String("econet.friendly_name", d.FriendlyName()),
attribute.String("econet.serial", d.SerialNumber),
attribute.String("econet.device_id", d.DeviceID),
attribute.String("econet.friendly_name", d.FriendlyName),
}
}
func infoAttrs(d *rheemcloud.Device) []attribute.KeyValue {
func infoAttrs(d *econetclient.Device) []attribute.KeyValue {
return append(deviceAttrs(d),
attribute.String("econet.type", d.Type().String()),
attribute.String("econet.generic_type", d.GenericType()),
attribute.String("econet.mode", d.Mode().String()),
attribute.String("econet.type", d.Type),
attribute.String("econet.generic_type", d.GenericType),
attribute.String("econet.mode", d.Mode),
)
}
+10 -89
View File
@@ -1,123 +1,44 @@
// Package econetmetrics registers OTEL observable gauges for Rheem EcoNet
// device state and periodically polls energy/water usage history so the
// values are available at Prometheus scrape time.
// device state. Values are read at scrape time from the econetclient snapshot,
// which a background poller keeps fresh over REST.
package econetmetrics
import (
"context"
"sync"
"time"
rheemcloud "github.com/kevinburke/rheemcloud-go"
"github.com/rs/zerolog"
"go.opentelemetry.io/otel/metric"
"gitea.libretechconsulting.com/rmcguire/go-app/pkg/otel"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/config"
"gitea.libretechconsulting.com/rmcguire/econet-exporter/pkg/econet/econetclient"
)
// usage holds the most recent polled energy/water totals for a device.
type usage struct {
energyKWH float64
energyType string
waterGallons float64
}
type Collector struct {
ctx context.Context
cfg *config.ServiceConfig
log *zerolog.Logger
client *rheemcloud.Client
client *econetclient.Client
meter metric.Meter
mu sync.RWMutex
usage map[string]usage // keyed by serial number
stop chan struct{}
}
func NewCollector(ctx context.Context, cfg *config.ServiceConfig, client *rheemcloud.Client) *Collector {
func NewCollector(ctx context.Context, cfg *config.ServiceConfig, client *econetclient.Client) *Collector {
return &Collector{
ctx: ctx,
cfg: cfg,
log: zerolog.Ctx(ctx),
client: client,
meter: otel.GetMeter(ctx, "econet"),
usage: make(map[string]usage),
stop: make(chan struct{}),
}
}
// Start registers the gauge callbacks and launches the usage poller and
// event drain goroutines. The rheemcloud client keeps live state fresh over
// MQTT; draining its event channel keeps the cap-64 buffer from filling.
func (c *Collector) Start() error {
// RegisterGauges registers the observable gauge callbacks. It is safe to call
// before the first refresh: the callbacks observe nothing while no devices are
// loaded, so /metrics comes up immediately.
func (c *Collector) RegisterGauges() error {
if err := c.registerLive(); err != nil {
return err
}
if err := c.registerUsage(); err != nil {
return err
}
go c.drainEvents()
go c.pollUsage()
return nil
}
func (c *Collector) Stop() { close(c.stop) }
func (c *Collector) drainEvents() {
events := c.client.Subscribe()
for {
select {
case <-c.stop:
return
case <-c.ctx.Done():
return
case ev, ok := <-events:
if !ok {
return
}
c.log.Debug().Str("event", ev.Kind.String()).Msg("econet event")
}
}
}
func (c *Collector) pollUsage() {
c.refreshUsage()
t := time.NewTicker(c.cfg.GetUsageInterval())
defer t.Stop()
for {
select {
case <-c.stop:
return
case <-c.ctx.Done():
return
case <-t.C:
c.refreshUsage()
}
}
}
// refreshUsage polls each device's energy/water usage for the current day
// and caches the totals for the usage gauge callback to read.
func (c *Collector) refreshUsage() {
now := time.Now()
start := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
for serial, d := range c.client.Devices() {
u := usage{}
if e, err := c.client.EnergyUsage(c.ctx, d, start, now); err == nil {
u.energyKWH, u.energyType = e.Total, e.EnergyType
} else {
c.log.Debug().Err(err).Str("serial", serial).Msg("energy usage poll failed")
}
if w, err := c.client.WaterUsage(c.ctx, d, start, now); err == nil {
u.waterGallons = w.Total
} else {
c.log.Debug().Err(err).Str("serial", serial).Msg("water usage poll failed")
}
c.mu.Lock()
c.usage[serial] = u
c.mu.Unlock()
}
return c.registerUsage()
}