Vehicles: add streaming feature (#24518)

api/feature.go

@@ -11,5 +11,6 @@ const (
 	Cacheable                // tariff
 	Offline                  // vehicle
 	Retryable                // vehicle
+	Streaming                // vehicle
 	WelcomeCharge            // vehicle
 )

api/feature_enumer.go

@@ -7,11 +7,11 @@ import (
 	"strings"
 )
 
-const _FeatureName = "CoarseCurrentIntegratedDeviceHeatingCacheableOfflineRetryableWelcomeCharge"
+const _FeatureName = "CoarseCurrentIntegratedDeviceHeatingCacheableOfflineRetryableStreamingWelcomeCharge"
 
-var _FeatureIndex = [...]uint8{0, 13, 29, 36, 45, 52, 61, 74}
+var _FeatureIndex = [...]uint8{0, 13, 29, 36, 45, 52, 61, 70, 83}
 
-const _FeatureLowerName = "coarsecurrentintegrateddeviceheatingcacheableofflineretryablewelcomecharge"
+const _FeatureLowerName = "coarsecurrentintegrateddeviceheatingcacheableofflineretryablestreamingwelcomecharge"
 
 func (i Feature) String() string {
 	i -= 1
@@ -31,10 +31,11 @@ func _FeatureNoOp() {
 	_ = x[Cacheable-(4)]
 	_ = x[Offline-(5)]
 	_ = x[Retryable-(6)]
-	_ = x[WelcomeCharge-(7)]
+	_ = x[Streaming-(7)]
+	_ = x[WelcomeCharge-(8)]
 }
 
-var _FeatureValues = []Feature{CoarseCurrent, IntegratedDevice, Heating, Cacheable, Offline, Retryable, WelcomeCharge}
+var _FeatureValues = []Feature{CoarseCurrent, IntegratedDevice, Heating, Cacheable, Offline, Retryable, Streaming, WelcomeCharge}
 
 var _FeatureNameToValueMap = map[string]Feature{
 	_FeatureName[0:13]:       CoarseCurrent,
@@ -49,8 +50,10 @@ var _FeatureNameToValueMap = map[string]Feature{
 	_FeatureLowerName[45:52]: Offline,
 	_FeatureName[52:61]:      Retryable,
 	_FeatureLowerName[52:61]: Retryable,
-	_FeatureName[61:74]:      WelcomeCharge,
-	_FeatureLowerName[61:74]: WelcomeCharge,
+	_FeatureName[61:70]:      Streaming,
+	_FeatureLowerName[61:70]: Streaming,
+	_FeatureName[70:83]:      WelcomeCharge,
+	_FeatureLowerName[70:83]: WelcomeCharge,
 }
 
 var _FeatureNames = []string{
@@ -60,7 +63,8 @@ var _FeatureNames = []string{
 	_FeatureName[36:45],
 	_FeatureName[45:52],
 	_FeatureName[52:61],
-	_FeatureName[61:74],
+	_FeatureName[61:70],
+	_FeatureName[70:83],
 }
 
 // FeatureString retrieves an enum value from the enum constants string name.

core/loadpoint.go

@@ -1701,7 +1701,7 @@ func (lp *Loadpoint) publishSocAndRange() {
 	}
 
 	// integrated device can bypass the update interval if vehicle is separately configured (legacy)
-	if lp.chargerHasFeature(api.IntegratedDevice) || lp.vehicleSocPollAllowed() {
+	if lp.chargerHasFeature(api.IntegratedDevice) || lp.vehicleHasFeature(api.Streaming) || lp.vehicleSocPollAllowed() {
 		lp.socUpdated = lp.clock.Now()
 
 		f, err := socEstimator.Soc(lp.GetChargedEnergy())

templates/definition/charger/semp-sma.yaml

@@ -8,7 +8,7 @@ products:
       generic: eCharger (SEMP)
 capabilities: ["mA", "1p3p"]
 requirements:
-  evcc: ["sponsorship"]
+  evcc: ["sponsorship", "skiptest"]
   description:
     en: |
       Configure the SEMP base URL (e.g. http://192.168.178.100/SEMP) and the device ID of the charger.

templates/definition/vehicle/cardata.yaml

@@ -56,4 +56,4 @@ render: |
   vin: {{ .vin }}
   clientid: {{ .clientid }}
   {{ include "vehicle-common" . }}
-  {{ include "vehicle-features" . }}
+  {{ include "vehicle-features" (set . "streaming" "true") }}

templates/definition/vehicle/homeassistant.yaml

@@ -87,9 +87,11 @@ params:
       en: Set charging current [A]
     example: "number.vehicle_charging_current"
     type: string
+  - preset: vehicle-features
 render: |
   type: homeassistant
   {{ include "vehicle-common" . }}
+  {{ include "vehicle-features" . }}
   uri: {{ .uri }}
   token: {{ .token }}
   sensors:

templates/definition/vehicle/mazda2mqtt.yaml

@@ -35,3 +35,4 @@ render: |
     source: mqtt
     topic: mazda2mqtt/{{ .vin }}/chargeInfo/drivingRangeKm
     timeout: {{ .timeout }}
+  features: ["streaming"]

util/templates/defaults.yaml

@@ -309,6 +309,14 @@ params:
     help:
       en: Vehicle supports 1A current steps only
       de: Fahrzeug unterstützt nur 1A Schritte der Ladestromvorgabe
+  - name: streaming
+    type: bool
+    description:
+      en: Supports streaming
+      de: Unterstützt Streaming
+    help:
+      en: Streaming data is received asynchronously
+      de: Streaming Datenempfang erfolgt asynchron
   - name: welcomecharge
     type: bool
     description:
@@ -505,6 +513,8 @@ presets:
     params:
       - name: coarsecurrent
         advanced: true
+      - name: streaming
+        advanced: true
       - name: welcomecharge
         advanced: true
   vehicle-language:

util/templates/includes/vehicle-features.tpl

@@ -7,5 +7,8 @@ features:
 {{- if eq .welcomecharge "true" }}
 - welcomecharge
 {{- end }}
+{{- if eq .streaming "true" }}
+- streaming
+{{- end }}
 {{- end }}
 {{- end }}

vehicle/bmw/cardata/mqtt.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"testing"
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
@@ -39,7 +40,9 @@ func NewMqttConnector(ctx context.Context, log *util.Logger, clientID string, ts
 		subscriptions: make(map[string]chan StreamingMessage),
 	}
 
-	go v.run(ctx, ts)
+	if !testing.Testing() {
+		go v.run(ctx, ts)
+	}
 
 	mqttConnections[clientID] = v
 
@@ -133,7 +136,7 @@ func (v *MqttConnector) runMqtt(ctx context.Context, token *oauth2.Token) error
 	return nil
 }
 
-func (v *MqttConnector) handler(c mqtt.Client, m mqtt.Message) {
+func (v *MqttConnector) handler(_ mqtt.Client, m mqtt.Message) {
 	var res StreamingMessage
 	if err := json.Unmarshal(m.Payload(), &res); err != nil {
 		v.log.ERROR.Println(m.Topic(), string(m.Payload()), err)

vehicle/bmw/cardata/provider.go

@@ -40,6 +40,7 @@ func NewProvider(ctx context.Context, log *util.Logger, api *API, ts oauth2.Toke
 	}
 
 	mqtt := NewMqttConnector(context.Background(), log, clientID, ts)
+	recvC := mqtt.Subscribe(vin)
 
 	go func() {
 		<-ctx.Done()
@@ -47,7 +48,7 @@ func NewProvider(ctx context.Context, log *util.Logger, api *API, ts oauth2.Toke
 	}()
 
 	go func() {
-		for msg := range mqtt.Subscribe(vin) {
+		for msg := range recvC {
 			v.mu.Lock()
 			maps.Copy(v.streaming, msg.Data)
 			v.mu.Unlock()

vehicle/bmw/cardata/provider_test.go

@@ -0,0 +1,46 @@
+package cardata
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/evcc-io/evcc/util"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/oauth2"
+)
+
+func TestCardataStreaming(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+
+	p := NewProvider(ctx, util.NewLogger("foo"), nil, oauth2.StaticTokenSource(&oauth2.Token{
+		AccessToken: "at",
+	}), "client", "vin")
+
+	keySoc := "vehicle.drivetrain.batteryManagement.header"
+	p.initial = map[string]TelematicDataPoint{
+		keySoc: {Value: "42"},
+	}
+
+	soc, err := p.Soc()
+	require.NoError(t, err)
+	require.Equal(t, 42.0, soc)
+
+	mqtt := mqttConnections["client"]
+	dataC := mqtt.subscriptions["vin"]
+	require.NotNil(t, dataC, "streaming channel")
+
+	dataC <- StreamingMessage{
+		Vin: "vin",
+		Data: map[string]StreamingData{
+			keySoc: {Value: "47"},
+		},
+	}
+
+	time.Sleep(100 * time.Millisecond)
+
+	soc, err = p.Soc()
+	require.NoError(t, err)
+	require.Equal(t, 47.0, soc)
+}