From 047f7c233d6d7fd8843304fc01728500360c6403 Mon Sep 17 00:00:00 2001 From: Ryan McGuire Date: Sun, 5 Jul 2026 16:59:53 -0400 Subject: [PATCH] remove rheemcloud pkg, implement against rest api only --- AGENTS.md | 37 ++-- README.md | 24 ++- config.yaml | 2 +- contrib/schema.json | 7 +- go.mod | 5 - go.sum | 10 - helm/values.yaml | 2 +- pkg/config/custom.go | 21 +- pkg/econet/econet.go | 74 ++++--- pkg/econet/econetclient/client.go | 324 ++++++++++++++++++++++++++++ pkg/econet/econetclient/device.go | 322 +++++++++++++++++++++++++++ pkg/econet/econetgrpc/device.go | 39 ++-- pkg/econet/econetgrpc/econet.go | 8 +- pkg/econet/econetmetrics/gauges.go | 81 +++---- pkg/econet/econetmetrics/metrics.go | 99 +-------- 15 files changed, 822 insertions(+), 233 deletions(-) create mode 100644 pkg/econet/econetclient/client.go create mode 100644 pkg/econet/econetclient/device.go diff --git a/AGENTS.md b/AGENTS.md index 6d34ffd..847ed31 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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) @@ -40,16 +51,16 @@ that's all it takes — no extra wiring. ## 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`). @@ -59,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) diff --git a/README.md b/README.md index f50dcc1..8f3bc03 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/config.yaml b/config.yaml index 9687d4c..1280969 100644 --- a/config.yaml +++ b/config.yaml @@ -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 diff --git a/contrib/schema.json b/contrib/schema.json index 2144cd2..ec8c896 100644 --- a/contrib/schema.json +++ b/contrib/schema.json @@ -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": { diff --git a/go.mod b/go.mod index fbeee0c..1595724 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index d406df9..5e56a0b 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/helm/values.yaml b/helm/values.yaml index 88072bf..e4dccea 100644 --- a/helm/values.yaml +++ b/helm/values.yaml @@ -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: diff --git a/pkg/config/custom.go b/pkg/config/custom.go index 06550d6..8498232 100644 --- a/pkg/config/custom.go +++ b/pkg/config/custom.go @@ -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 @@ -30,8 +30,9 @@ type ServiceConfig struct { // 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 @@ -50,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 } diff --git a/pkg/econet/econet.go b/pkg/econet/econet.go index e8eba0a..e8ca802 100644 --- a/pkg/econet/econet.go +++ b/pkg/econet/econet.go @@ -5,17 +5,16 @@ package econet import ( "context" - "crypto/tls" "fmt" - "log/slog" - "net/http" + "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" @@ -25,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 @@ -34,46 +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(), - HTTPClient: &http.Client{ - Transport: &http.Transport{ - TLSClientConfig: &tls.Config{ - InsecureSkipVerify: cfg.EconetTLSInsecure, - }, - }, - }, - }) - 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 { @@ -95,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 }, diff --git a/pkg/econet/econetclient/client.go b/pkg/econet/econetclient/client.go new file mode 100644 index 0000000..2c476fa --- /dev/null +++ b/pkg/econet/econetclient/client.go @@ -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" +} diff --git a/pkg/econet/econetclient/device.go b/pkg/econet/econetclient/device.go new file mode 100644 index 0000000..325cc6d --- /dev/null +++ b/pkg/econet/econetclient/device.go @@ -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 +} diff --git a/pkg/econet/econetgrpc/device.go b/pkg/econet/econetgrpc/device.go index a4bb28c..ba769ed 100644 --- a/pkg/econet/econetgrpc/device.go +++ b/pkg/econet/econetgrpc/device.go @@ -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), } } diff --git a/pkg/econet/econetgrpc/econet.go b/pkg/econet/econetgrpc/econet.go index 516da5a..d9e324a 100644 --- a/pkg/econet/econetgrpc/econet.go +++ b/pkg/econet/econetgrpc/econet.go @@ -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, diff --git a/pkg/econet/econetmetrics/gauges.go b/pkg/econet/econetmetrics/gauges.go index e29ea28..e732710 100644 --- a/pkg/econet/econetmetrics/gauges.go +++ b/pkg/econet/econetmetrics/gauges.go @@ -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), ) } diff --git a/pkg/econet/econetmetrics/metrics.go b/pkg/econet/econetmetrics/metrics.go index 3292c87..22711e2 100644 --- a/pkg/econet/econetmetrics/metrics.go +++ b/pkg/econet/econetmetrics/metrics.go @@ -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() }