generated from rmcguire/go-server-with-otel
Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 38078a330e | |||
| 047f7c233d | |||
| dcdf5e4230 | |||
| 1b42d26ffc | |||
| a8e0a6ef1e | |||
| 059b89b5cd | |||
| 87e53bd620 | |||
| c481d3f032 |
@@ -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
@@ -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.
|
||||
|
||||
@@ -19,6 +19,8 @@ FROM alpine:latest
|
||||
|
||||
ARG APP_NAME=econet-exporter
|
||||
|
||||
RUN apk add --no-cache ca-certificates
|
||||
|
||||
WORKDIR /app
|
||||
USER 100:101
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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
@@ -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
@@ -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": {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
@@ -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.1.0"
|
||||
appVersion: "v0.3.0"
|
||||
|
||||
dependencies:
|
||||
- name: hull
|
||||
|
||||
+1
-1
@@ -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
@@ -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
@@ -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
|
||||
},
|
||||
|
||||
@@ -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"
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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),
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user