remove rheemcloud pkg, implement against rest api only

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
 }

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
 		},

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"
+}

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
+}

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),
 	}
 }

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,

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),
 	)
 }
 

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()
 }