package meter import ( "context" "encoding/json" "fmt" "net/http" "runtime/debug" "strings" "time" "github.com/cenkalti/backoff/v4" "github.com/evcc-io/evcc/api" "github.com/evcc-io/evcc/meter/tibber" "github.com/evcc-io/evcc/util" "github.com/evcc-io/evcc/util/request" "github.com/evcc-io/evcc/util/transport" "github.com/hasura/go-graphql-client" ) func init() { registry.AddCtx("tibber-pulse", NewTibberFromConfig) } func getUserAgent() string { graphqlClientVersion := "unknown" if info, ok := debug.ReadBuildInfo(); ok { for _, dep := range info.Deps { if dep.Path == "github.com/hasura/go-graphql-client" { graphqlClientVersion = baseVersion(dep.Version) } } } return fmt.Sprintf("evcc/%s hasura/go-graphql-client/%s", util.FormattedVersion(), graphqlClientVersion) } func baseVersion(v string) string { if i := strings.IndexAny(v, "-+"); i != -1 { return v[:i] } return v } type Tibber struct { data *util.Monitor[tibber.LiveMeasurement] } func NewTibberFromConfig(ctx context.Context, other map[string]any) (api.Meter, error) { cc := struct { Token string HomeID string Timeout time.Duration }{ Timeout: 2 * time.Minute, } if err := util.DecodeOther(other, &cc); err != nil { return nil, err } if cc.Token == "" { return nil, api.ErrMissingToken } log := util.NewLogger("pulse").Redact(cc.Token, cc.HomeID) // query client qclient := tibber.NewClient(log, cc.Token) if cc.HomeID == "" { home, err := qclient.DefaultHome("") if err != nil { return nil, err } cc.HomeID = home.ID } var res struct { Viewer struct { WebsocketSubscriptionUrl string } } ctx2, cancel := context.WithTimeout(ctx, request.Timeout) defer cancel() if err := qclient.Query(ctx2, &res, nil); err != nil { return nil, err } t := &Tibber{ data: util.NewMonitor[tibber.LiveMeasurement](cc.Timeout), } // subscription client client := graphql.NewSubscriptionClient(res.Viewer.WebsocketSubscriptionUrl). WithProtocol(graphql.GraphQLWS). WithWebSocketOptions(graphql.WebsocketOptions{ HTTPClient: &http.Client{ Transport: &transport.Decorator{ Base: http.DefaultTransport, Decorator: transport.DecorateHeaders(map[string]string{ "User-Agent": getUserAgent(), }), }, }, }). WithConnectionParams(map[string]any{ "token": cc.Token, }). WithRetryTimeout(20 * time.Second). // 2 tries, then exit to outer retry loop that has backoff WithRetryDelay(10 * time.Second). WithWriteTimeout(request.Timeout). WithReadTimeout(90 * time.Second). WithLog(log.TRACE.Println). OnConnected(func() { log.DEBUG.Println("websocket connected") }). OnDisconnected(func() { log.WARN.Println("websocket disconnected") }). OnSubscriptionComplete(func(_ graphql.Subscription) { log.DEBUG.Println("websocket subscription completed by server") }). OnError(func(sc *graphql.SubscriptionClient, err error) error { // Don't let graphql client reconnect when authorization fails if sc.IsUnauthorized(err) { return err } // Don't let Hasura go graphql client reconnect when too many initialisation requests // Reconnection will be attempted in the loop later if sc.IsTooManyInitialisationRequests(err) { return err } log.ERROR.Printf("error occurred: %v", err) return nil }) done := make(chan error, 1) go func(done chan error) { done <- t.subscribe(client, cc.HomeID, log) }(done) select { case err := <-done: if err != nil { return nil, err } case <-time.After(cc.Timeout): return nil, api.ErrTimeout } go func() { <-ctx.Done() log.INFO.Println("context canceled, closing Tibber Pulse client") if err := client.Close(); err != nil { log.ERROR.Printf("error closing Tibber Pulse client: %v", err) } }() go func() { // The pulse sometimes declines valid(!) subscription requests, and asks the client to disconnect. // Therefore we need to restart the client when exiting gracefully upon server request // https://github.com/evcc-io/evcc/issues/17925#issuecomment-2621458890 // Note that there are several reconnection strategies in play: // 1. Mechanism built into Hasura go graphql client // 2. This loop, which is triggered server when Hasura exits on error or gracefully // 3. evcc itself restarts if the client exits with an error var reconnectCount int bo := backoff.NewExponentialBackOff( backoff.WithInitialInterval(30*time.Second), backoff.WithMaxInterval(10*time.Minute), backoff.WithMaxElapsedTime(0), ) for { reconnectCount++ log.DEBUG.Printf("graphql client connection attempt #%d", reconnectCount) startTime := time.Now() err := client.Run() duration := time.Since(startTime).Round(time.Second) if err != nil { log.ERROR.Printf("graphql client exited with error at %s: %v", duration, err) // Do not retry if unauthorized if client.IsUnauthorized(err) { log.ERROR.Println("Not retrying due to unauthorized error.") return } } else { log.DEBUG.Printf("graphql client exited gracefully at %s", duration) bo.Reset() } delay := bo.NextBackOff() select { case <-time.After(delay): log.DEBUG.Printf("reconnecting after %v backoff delay", delay) case <-ctx.Done(): return } } }() log.DEBUG.Printf("!! User-Agent set to %s", getUserAgent()) return t, nil } func (t *Tibber) subscribe(client *graphql.SubscriptionClient, homeID string, log *util.Logger) error { var query struct { tibber.LiveMeasurement `graphql:"liveMeasurement(homeId: $homeId)"` } _, err := client.Subscribe(&query, map[string]any{ "homeId": graphql.ID(homeID), }, func(data []byte, err error) error { if err != nil { log.ERROR.Printf("Error during subscription: %v", err) return err } var res struct { LiveMeasurement tibber.LiveMeasurement } if err := json.Unmarshal(data, &res); err != nil { log.ERROR.Printf("Error unmarshaling data: %v", err) return err } t.data.Set(res.LiveMeasurement) return nil }) return err } func (t *Tibber) CurrentPower() (float64, error) { res, err := t.data.Get() if err != nil { return 0, err } return res.Power - res.PowerProduction, nil } var _ api.PhaseCurrents = (*Tibber)(nil) // Currents implements the api.PhaseCurrents interface func (t *Tibber) Currents() (float64, float64, float64, error) { res, err := t.data.Get() if err != nil { return 0, 0, 0, err } return res.CurrentL1, res.CurrentL2, res.CurrentL3, nil }