Repository: evcc-io/evcc
Pull request #1963 (2021-12-01 10:12:40)

Add OCPP charger
Refs https://github.com/evcc-io/evcc/discussions/1042#discussioncomment-1675658

Todos:

- [x] match `ConnectorID`
- [x] understand `idTags`
- [x] verify vehicle status A vs B
- [x] verify smart charging behaviour
- [ ] detect charger capabilities (smart charging? meter? phases?)
- [ ] add meter support (separate PR)
- [ ] support boxes with >1 connectors (separate PR due to connector id 0 handling complexity)

/cc @xantalor @utsavanand2 @daviid

abierfert: Hallo in die Runde!

Ich kenn mich zwar mit GO nicht aus, kann aber gerne beim testen helfen. Hab eine EVBox Elvi und einen ZOE. Kein PV.

Ich hab den Branch ausgecheckt und gebaut und installiert. Stellt sich nun die Frage wie eine Beispielkonfiguration aussehen könnte?
andig: Guckst du hier: https://github.com/evcc-io/evcc/pull/1963/files#diff-428bcc3f4e7da95e0dbd70c679c4cd4c4f672004721185ee58131593bfa12d92R36
xantalor: > Hallo in die Runde!
> 
> Ich kenn mich zwar mit GO nicht aus, kann aber gerne beim testen helfen. Hab eine EVBox Elvi und einen ZOE. Kein PV.
> 
> Ich hab den Branch ausgecheckt und gebaut und installiert. Stellt sich nun die Frage wie eine Beispielkonfiguration aussehen könnte?

```

chargers: 
- name: Elvi
  type: ocpp
  stationid: EVB-P....
  idtag: .... #id vom lade token
  connector: 1
  meterSupported: true # wenn vorhanden bei deiner Elvi
```

Patch:
diff --git a/charger/ocpp.go b/charger/ocpp.go
new file mode 100644
index 00000000..33117c76
--- /dev/null
+++ b/charger/ocpp.go
@@ -0,0 +1,358 @@
+package charger
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/evcc-io/evcc/api"
+	"github.com/evcc-io/evcc/charger/ocpp"
+	"github.com/evcc-io/evcc/util"
+	"github.com/evcc-io/evcc/util/request"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/smartcharging"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/types"
+)
+
+// OCPP charger implementation
+type OCPP struct {
+	log                     *util.Logger
+	cp                      *ocpp.CP
+	id                      string
+	connector               int
+	idtag                   string
+	phases                  int
+	current                 float64
+	phaseSwitchingSupported bool
+}
+
+func init() {
+	registry.Add("ocpp", NewOCPPFromConfig)
+}
+
+// NewOCPPFromConfig creates a OCPP charger from generic config
+func NewOCPPFromConfig(other map[string]interface{}) (api.Charger, error) {
+	cc := struct {
+		StationId      string
+		IdTag          string
+		Connector      int
+		MeterSupported bool
+		MeterInterval  time.Duration
+		InitialReset   core.ResetType
+	}{
+		Connector:     1,
+		MeterInterval: 0, // disable meter sample data as we are taking care on our own
+	}
+
+	if err := util.DecodeOther(other, &cc); err != nil {
+		return nil, err
+	}
+
+	switch cc.InitialReset {
+	case
+		"",
+		core.ResetTypeSoft,
+		core.ResetTypeHard:
+	default:
+		return nil, fmt.Errorf("unknown configuration option detected for reset: %s", cc.InitialReset)
+	}
+
+	ocpp, err := NewOCPP(cc.StationId, cc.Connector, cc.IdTag, cc.MeterSupported, cc.MeterInterval, cc.InitialReset)
+	if err != nil {
+		return ocpp, err
+	}
+
+	var (
+		meter        func() (float64, error)
+		meterCurrent func() (float64, float64, float64, error)
+		chargeRater  func() (float64, error)
+	)
+
+	if cc.MeterSupported {
+		meter = ocpp.currentPower
+		meterCurrent = ocpp.currents
+		chargeRater = ocpp.chargedEnergy
+	}
+
+	return decorateOCPP(ocpp, meter, meterCurrent, chargeRater), nil
+}
+
+//go:generate go run ../cmd/tools/decorate.go -f decorateOCPP -b *OCPP -r api.Charger -t "api.Meter,CurrentPower,func() (float64, error)" -t "api.MeterCurrent,Currents,func() (float64, float64, float64, error)" -t "api.ChargeRater,ChargedEnergy,func() (float64, error)"
+
+// NewOCPP creates OCPP charger
+func NewOCPP(id string, connector int, idtag string, meterSupported bool, meterInterval time.Duration, initialReset core.ResetType) (*OCPP, error) {
+	cp := ocpp.Instance().Register(id, meterSupported)
+	c := &OCPP{
+		log:       util.NewLogger(fmt.Sprintf("ocpp-%s:%d", id, connector)),
+		cp:        cp,
+		id:        id,
+		connector: connector,
+		idtag:     idtag,
+	}
+
+	if err := cp.Boot(); err != nil {
+		return nil, err
+	}
+
+	var (
+		rc                           = make(chan error, 1)
+		options                      []core.ConfigurationKey
+		meterValuesSampledDataString string
+		meterValuesSampleInterval    string
+	)
+
+	err := ocpp.Instance().CS().GetConfiguration(id, func(resp *core.GetConfigurationConfirmation, err error) {
+		options = resp.ConfigurationKey
+
+		for _, opt := range options {
+			c.log.TRACE.Printf("%s (%t): %s", opt.Key, opt.Readonly, *opt.Value)
+			switch opt.Key {
+			case ocpp.KeyMeterValuesSampledData:
+				meterValuesSampledDataString = *opt.Value
+			case ocpp.KeyMeterValueSampleInterval:
+				meterValuesSampleInterval = *opt.Value
+			case string(ocpp.KeyConnectorSwitch3to1PhaseSupported):
+				// Detection of 1 phase charging/switching support
+				b, err := strconv.ParseBool(*opt.Value)
+				if err != nil {
+					rc <- err
+					return
+				}
+
+				c.phaseSwitchingSupported = b
+			}
+		}
+
+		rc <- err
+	}, []string{})
+
+	if err := c.wait(err, rc); err != nil {
+		return nil, err
+	}
+
+	if err := cp.DetectCapabilities(options); err != nil {
+		return nil, err
+	}
+
+	{ // Check supported connectors of charge point
+		supported := cp.GetNumberOfSupportedConnectors()
+		if c.connector > supported {
+			return nil, fmt.Errorf("configured connector is not available, max available connectors %d", supported)
+		}
+	}
+
+	if meterSupported {
+		if meterValuesSampledDataString != "Current.Import,Current.Offered,Energy.Active.Import.Register,Power.Active.Import,Temperature" {
+			c.log.TRACE.Printf("Current values \n\t%s != \n\t%+v", ocpp.ValuePreferedMeterValuesSampleData, meterValuesSampledDataString)
+			rc = make(chan error, 1)
+			err = ocpp.Instance().CS().ChangeConfiguration(id, func(resp *core.ChangeConfigurationConfirmation, err error) {
+				c.log.TRACE.Printf("ChangeMeterConfigurationRequest %T: %+v", resp, resp)
+
+				if resp.Status == core.ConfigurationStatusRejected {
+					rc <- fmt.Errorf("configuration change rejected")
+				}
+
+				rc <- err
+			}, ocpp.KeyMeterValuesSampledData, ocpp.ValuePreferedMeterValuesSampleData)
+
+			if err := c.wait(err, rc); err != nil {
+				return nil, err
+			}
+		}
+
+		{
+			intervalStr := fmt.Sprintf("%d", int(meterInterval.Seconds()))
+			if meterValuesSampleInterval != intervalStr {
+				rc = make(chan error, 1)
+
+				err := ocpp.Instance().CS().ChangeConfiguration(id, func(resp *core.ChangeConfigurationConfirmation, err error) {
+					c.log.TRACE.Printf("ChangeSampleMeterValueInterval %T: %v", resp, resp)
+
+					if resp.Status == core.ConfigurationStatusRejected {
+						rc <- fmt.Errorf("configuration of meterinterval rejected: %w", err)
+					}
+					rc <- err
+				}, ocpp.KeyMeterValueSampleInterval, intervalStr)
+
+				if err := c.wait(err, rc); err != nil {
+					return nil, err
+				}
+			}
+		}
+
+		// get inital meter values
+		if meterSupported {
+			ocpp.Instance().TriggerMeterValueRequest(cp)
+		}
+	}
+
+	if initialReset != "" {
+		t := core.ResetTypeSoft
+		if initialReset == core.ResetTypeHard {
+			t = core.ResetTypeHard
+		}
+
+		ocpp.Instance().TriggerResetRequest(cp, t)
+	}
+
+	// TODO: check for running transaction
+
+	return c, nil
+}
+
+// Enabled implements the api.Charger interface
+func (c *OCPP) Enabled() (bool, error) {
+	current, err := c.cp.Status()
+	if current == api.StatusC {
+		return true, err
+	}
+
+	return false, err
+}
+
+func (c *OCPP) wait(err error, rc chan error) error {
+	if err == nil {
+		select {
+		case err = <-rc:
+			close(rc)
+		case <-time.After(request.Timeout):
+			err = api.ErrTimeout
+		}
+	}
+	return err
+}
+
+// Enable implements the api.Charger interface
+func (c *OCPP) Enable(enable bool) error {
+	var err error
+	rc := make(chan error, 1)
+
+	if enable {
+		err = ocpp.Instance().CS().RemoteStartTransaction(c.id, func(resp *core.RemoteStartTransactionConfirmation, err error) {
+			c.log.TRACE.Printf("RemoteStartTransaction %T: %+v", resp, resp)
+
+			if err == nil && resp != nil && resp.Status != types.RemoteStartStopStatusAccepted {
+				err = errors.New(string(resp.Status))
+			}
+
+			rc <- err
+		}, c.idtag, func(request *core.RemoteStartTransactionRequest) {
+			request.ConnectorId = &c.connector
+		})
+	} else {
+		err = ocpp.Instance().CS().RemoteStopTransaction(c.id, func(resp *core.RemoteStopTransactionConfirmation, err error) {
+			c.log.TRACE.Printf("RemoteStopTransaction %T: %+v", resp, resp)
+
+			if err == nil && resp != nil && resp.Status != types.RemoteStartStopStatusAccepted {
+				err = errors.New(string(resp.Status))
+			}
+
+			rc <- err
+		}, c.cp.TransactionID())
+	}
+
+	return c.wait(err, rc)
+}
+
+func (c *OCPP) setChargingProfile(connectorid int, profile *types.ChargingProfile) error {
+	c.log.TRACE.Printf("SetChargingProfileRequest %T: %+v", profile, profile)
+	c.log.TRACE.Printf("SetChargingProfileRequest %T: %+v", profile.ChargingSchedule, profile.ChargingSchedule)
+
+	rc := make(chan error, 1)
+	err := ocpp.Instance().CS().SetChargingProfile(c.id, func(resp *smartcharging.SetChargingProfileConfirmation, err error) {
+		c.log.TRACE.Printf("SetChargingProfileResponse %T: %+v", resp, resp)
+		if err == nil && resp != nil && resp.Status != smartcharging.ChargingProfileStatusAccepted {
+			err = errors.New(string(resp.Status))
+		}
+
+		rc <- err
+	}, connectorid, profile)
+
+	return c.wait(err, rc)
+}
+
+// setPeriod sets a single charging schedule period with given current and phases
+func (c *OCPP) setPeriod(current float64, phases int) error {
+	period := types.NewChargingSchedulePeriod(0, current)
+
+	c.log.TRACE.Printf("current phases: %d, current current: %f", phases, current)
+	if phases > 0 {
+		period.NumberPhases = &phases
+	}
+
+	// connectorID: 0 - profile will be applied to all connectors
+	err := c.setChargingProfile(0, getMaxCharginProfile(period))
+	if err != nil {
+		c.log.TRACE.Printf("failed to set charging profile: %s", err)
+	}
+
+	return err
+}
+
+func getMaxCharginProfile(period types.ChargingSchedulePeriod) *types.ChargingProfile {
+	return &types.ChargingProfile{
+		ChargingProfileId:      1,
+		StackLevel:             1,
+		ChargingProfilePurpose: types.ChargingProfilePurposeChargePointMaxProfile,
+		ChargingProfileKind:    types.ChargingProfileKindAbsolute,
+		ChargingSchedule: &types.ChargingSchedule{
+			StartSchedule:          types.NewDateTime(time.Now().Add(-1 * time.Hour)),
+			ChargingRateUnit:       types.ChargingRateUnitAmperes,
+			ChargingSchedulePeriod: []types.ChargingSchedulePeriod{period},
+		},
+	}
+}
+
+// MaxCurrent implements the api.Charger interface
+func (c *OCPP) MaxCurrent(current int64) error {
+	return c.MaxCurrentMillis(float64(current))
+}
+
+// MaxCurrentMillis implements the api.ChargerEx interface
+func (c *OCPP) MaxCurrentMillis(current float64) error {
+	err := c.setPeriod(current, c.phases)
+	if err == nil {
+		c.current = current
+	}
+	return err
+}
+
+// Status implements the api.Charger interface
+func (c *OCPP) Status() (api.ChargeStatus, error) {
+	return c.cp.Status()
+}
+
+// TODO: Phases1p3p implements the api.ChargePhases interface
+// func (c *OCPP) Phases1p3p(phases int) error {
+// 	if !c.phaseSwitchingSupported {
+// 		return fmt.Errorf("phase switching is not supported by the charger")
+// 	}
+
+// 	err := c.setPeriod(c.current, phases)
+// 	if err == nil {
+// 		c.phases = phases
+// 	}
+
+// 	return err
+// }
+
+// CurrentPower implements the api.Meter interface
+func (c *OCPP) currentPower() (float64, error) {
+	return c.cp.CurrentPower()
+}
+
+// ChargedEnergy implements the api.ChargeRater interface
+func (c *OCPP) chargedEnergy() (float64, error) {
+	return c.cp.ChargedEnergy()
+}
+
+// Currents implements the api.MeterCurrent interface
+func (c *OCPP) currents() (float64, float64, float64, error) {
+	return c.cp.Currents()
+}
+
+// // Identify implements the api.Identifier interface
+// func (c *OCPP) Identify() (string, error) {
+// 	return "", errors.New("not implemented")
+// }
diff --git a/charger/ocpp/cp.go b/charger/ocpp/cp.go
new file mode 100644
index 00000000..d795d699
--- /dev/null
+++ b/charger/ocpp/cp.go
@@ -0,0 +1,315 @@
+package ocpp
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/evcc-io/evcc/api"
+	"github.com/evcc-io/evcc/util"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/types"
+)
+
+const timeout = 2 * time.Minute
+
+// Meter Profile Key
+const (
+	KeyMeterValuesSampledData   = "MeterValuesSampledData"
+	KeyMeterValueSampleInterval = "MeterValueSampleInterval"
+)
+
+// TODO: Maybe move this to the config?
+const ValuePreferedMeterValuesSampleData = "Current.Import.L1,Current.Import.L2,Current.Import.L3,Current.Offered,Energy.Active.Import.Register,Power.Active.Import,Temperature"
+
+type SmartchargingChargeProfileKey string
+
+// Smart Charging Profile Key
+const (
+	KeyChargeProfileMaxStackLevel              SmartchargingChargeProfileKey = "ChargeProfileMaxStackLevel"
+	KeyChargingScheduleAllowedChargingRateUnit SmartchargingChargeProfileKey = "ChargingScheduleAllowedChargingRateUnit"
+	KeyChargingScheduleMaxPeriods              SmartchargingChargeProfileKey = "ChargingScheduleMaxPeriods"
+	KeyConnectorSwitch3to1PhaseSupported       SmartchargingChargeProfileKey = "ConnectorSwitch3to1PhaseSupported"
+	KeyMaxChargingProfilesInstalled            SmartchargingChargeProfileKey = "MaxChargingProfilesInstalled"
+)
+
+type smartChargingProfile struct {
+	// Max StackLevel of a ChargingProfile. The number defined also indicates the max allowed
+	// number of installed charging scheduls per Charging Profile Purpose
+	ChargeProfileMaxStackLevel int
+	// A list of supported quantities for use in a ChargingSchedule.
+	// Allowed values: 'Current' and 'Power'
+	ChargingScheduleAllowedChargingRateUnit []string
+	// Maximum number of periods that may be defined per ChargingSchedule
+	ChargingScheduleMaxPeriods int
+	// Defines if this Charge Point support switching from 3 to 1 phase during a charging session.
+	ConnectorSwitch3to1PhaseSupported bool
+	// Maximum number of Charging profiles instsalled at a time.
+	MaxChargingProfilesInstalled int
+}
+
+type CP struct {
+	mu  sync.Mutex
+	log *util.Logger
+	id  string
+
+	updated     time.Time
+	initialized *sync.Cond
+	boot        *core.BootNotificationRequest
+	status      *core.StatusNotificationRequest
+
+	meterSupported            bool
+	measureDoneCh             chan struct{}
+	latestMeterValueTimestamp time.Time
+	measureands               map[string]types.SampledValue
+	meterTrickerRunning       bool
+
+	supportedNumberOfConnectors int
+	smartChargingCapabilities   smartChargingProfile
+
+	currentTransaction Transaction
+}
+
+func (cp *CP) DetectCapabilities(opts []core.ConfigurationKey) error {
+	options := make(map[string]core.ConfigurationKey)
+	for _, opt := range opts {
+		options[opt.Key] = opt
+	}
+
+	{
+		supported, err := parseIntOption("NumberOfConnectors", options)
+		if err != nil {
+			return err
+		}
+
+		cp.supportedNumberOfConnectors = supported
+	}
+
+	smartChargingCapabilities, err := detectSmartChargingCapabilities(options)
+	if err != nil {
+		return err
+	}
+
+	cp.smartChargingCapabilities = smartChargingCapabilities
+
+	return nil
+}
+
+func (cp *CP) GetNumberOfSupportedConnectors() int {
+	return cp.supportedNumberOfConnectors
+}
+
+func detectSmartChargingCapabilities(options map[string]core.ConfigurationKey) (smartChargingProfile, error) {
+	var profile smartChargingProfile
+
+	{ // required
+		val, err := parseIntOption(KeyChargeProfileMaxStackLevel, options)
+		if err != nil {
+			return profile, err
+		}
+
+		profile.ChargeProfileMaxStackLevel = val
+	}
+
+	{ // required
+		val, err := parseIntOption(KeyChargingScheduleMaxPeriods, options)
+		if err != nil {
+			return profile, err
+		}
+
+		profile.ChargingScheduleMaxPeriods = val
+	}
+
+	{ // required
+		val, err := parseIntOption(KeyMaxChargingProfilesInstalled, options)
+		if err != nil {
+			return profile, err
+		}
+
+		profile.MaxChargingProfilesInstalled = val
+	}
+
+	{ // required
+		opt, found := options[string(KeyChargingScheduleAllowedChargingRateUnit)]
+		if !found || opt.Value == nil {
+			return profile, fmt.Errorf("smart charging key '%s' not found", KeyChargingScheduleAllowedChargingRateUnit)
+		}
+
+		vals := strings.Split(*opt.Value, ",")
+		profile.ChargingScheduleAllowedChargingRateUnit = append(profile.ChargingScheduleAllowedChargingRateUnit, vals...)
+	}
+
+	{ // optional
+		var supported bool
+		opt, found := options[string(KeyConnectorSwitch3to1PhaseSupported)]
+		if found {
+			var err error
+			supported, err = strconv.ParseBool(*opt.Value)
+			if err != nil {
+				return profile, fmt.Errorf("invalid value for key: %s", opt.Key)
+			}
+		}
+
+		profile.ConnectorSwitch3to1PhaseSupported = supported
+	}
+
+	return profile, nil
+
+}
+
+func parseIntOption(key SmartchargingChargeProfileKey, options map[string]core.ConfigurationKey) (int, error) {
+	opt, found := options[string(key)]
+	if !found || opt.Value == nil {
+		return 0, fmt.Errorf("smart charging key '%s' not found", key)
+	}
+
+	val, err := strconv.Atoi(*opt.Value)
+	if err != nil {
+		return 0, fmt.Errorf("invalid value for key: %s", key)
+	}
+
+	return val, nil
+}
+
+// Boot waits for the CP to register itself
+func (cp *CP) Boot() error {
+	bootC := make(chan struct{})
+	go func() {
+		cp.mu.Lock()
+		defer cp.mu.Unlock()
+
+		for cp.boot == nil || cp.status == nil {
+			cp.initialized.Wait()
+		}
+
+		close(bootC)
+	}()
+
+	select {
+	case <-bootC:
+		cp.update()
+		return nil
+	case <-time.After(timeout):
+		return api.ErrTimeout
+	}
+}
+
+// TransactionID returns the current transaction id
+func (cp *CP) TransactionID() int {
+	cp.mu.Lock()
+	defer cp.mu.Unlock()
+
+	return cp.currentTransaction.ID
+}
+
+func (cp *CP) Status() (api.ChargeStatus, error) {
+	cp.mu.Lock()
+	defer cp.mu.Unlock()
+
+	res := api.StatusNone
+
+	cp.log.TRACE.Printf("last status update from CP: %s", cp.updated.Format(time.RFC3339))
+	cp.log.TRACE.Printf("current transaction ID: %d", cp.currentTransaction.ID)
+
+	if time.Since(cp.updated) > timeout {
+		return res, api.ErrTimeout
+	}
+
+	if cp.status.ErrorCode != core.NoError {
+		cp.log.DEBUG.Printf("chargepoint error: %s: %s", cp.status.ErrorCode, cp.status.Info)
+	}
+
+	switch cp.status.Status {
+	case core.ChargePointStatusAvailable, // "Available"
+		core.ChargePointStatusUnavailable: // "Unavailable"
+		res = api.StatusA
+	case
+		core.ChargePointStatusPreparing,     // "Preparing"
+		core.ChargePointStatusSuspendedEVSE, // "SuspendedEVSE"
+		core.ChargePointStatusSuspendedEV,   // "SuspendedEV"
+		core.ChargePointStatusFinishing:     // "Finishing"
+		res = api.StatusB
+	case core.ChargePointStatusCharging: // "Charging"
+		res = api.StatusC
+	case core.ChargePointStatusReserved, // "Reserved"
+		core.ChargePointStatusFaulted: // "Faulted"
+		return api.StatusF, fmt.Errorf("chargepoint status: %s", cp.status.ErrorCode)
+	default:
+		return api.StatusNone, fmt.Errorf("invalid chargepoint status: %s", cp.status.Status)
+	}
+
+	return res, nil
+}
+
+func (cp *CP) CurrentPower() (float64, error) {
+	cp.mu.Lock()
+	defer cp.mu.Unlock()
+
+	if power, ok := cp.measureands[string(types.MeasurandPowerActiveImport)]; ok {
+		return strconv.ParseFloat(power.Value, 64)
+	}
+
+	return 0, nil
+}
+
+// func (cp *CP) TotalEnergy() (float64, error) {
+// 	cp.mu.Lock()
+// 	defer cp.mu.Unlock()
+
+// 	// if energy, ok := cp.measureands[string(types.MeasurandEnergyActiveImportRegister)]; ok {
+// 	// 	v, err := strconv.ParseInt(energy.Value, 10, 64)
+// 	// 	if err != nil {
+// 	// 		return 0, err
+// 	// 	}
+
+// 	return float64(cp.currentTransaction.Charged) / 1000, nil
+
+// 	// loaded := float64(int(v)-cp.currentTransaction.MeterValueStart) / 1000
+
+// 	// return loaded, nil
+// 	// }
+
+// 	// return 0, nil
+// }
+
+func (cp *CP) ChargedEnergy() (float64, error) {
+	cp.mu.Lock()
+	defer cp.mu.Unlock()
+
+	return float64(cp.currentTransaction.Charged) / 1000, nil
+}
+
+func getKeyCurrentPhase(phase int) string {
+	return string(types.MeasurandCurrentImport) + "@L" + strconv.Itoa(phase)
+}
+
+func (cp *CP) Currents() (float64, float64, float64, error) {
+	cp.mu.Lock()
+	defer cp.mu.Unlock()
+
+	var (
+		currents = make(map[int]float64)
+
+		err error
+	)
+
+	for _, phase := range []int{1, 2, 3} {
+		if current, ok := cp.measureands[getKeyCurrentPhase(phase)]; ok {
+			currents[phase], err = strconv.ParseFloat(current.Value, 64)
+			if err != nil {
+				return 0, 0, 0, fmt.Errorf("invalid current for phase %d: %w", phase, err)
+			}
+		}
+	}
+
+	return currents[1], currents[2], currents[3], nil
+}
+
+func (cp *CP) MeterSupported() bool {
+	cp.mu.Lock()
+	defer cp.mu.Unlock()
+
+	return cp.meterSupported
+}
diff --git a/charger/ocpp/cp_core.go b/charger/ocpp/cp_core.go
new file mode 100644
index 00000000..ec517973
--- /dev/null
+++ b/charger/ocpp/cp_core.go
@@ -0,0 +1,225 @@
+package ocpp
+
+import (
+	"strconv"
+	"time"
+
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/firmware"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/types"
+)
+
+func (cp *CP) Authorize(request *core.AuthorizeRequest) (*core.AuthorizeConfirmation, error) {
+	cp.log.TRACE.Printf("%T: %+v", request, request)
+
+	// TODO check if this authorizes foreign RFID tags
+	res := &core.AuthorizeConfirmation{
+		IdTagInfo: &types.IdTagInfo{
+			Status: types.AuthorizationStatusAccepted,
+		},
+	}
+
+	return res, nil
+}
+
+func (cp *CP) BootNotification(request *core.BootNotificationRequest) (*core.BootNotificationConfirmation, error) {
+	cp.log.TRACE.Printf("%T: %+v", request, request)
+
+	if request != nil {
+		cp.mu.Lock()
+		defer cp.mu.Unlock()
+
+		cp.boot = request
+		cp.initialized.Broadcast()
+	}
+
+	res := &core.BootNotificationConfirmation{
+		CurrentTime: types.NewDateTime(time.Now()),
+		Interval:    60, // TODO
+		Status:      core.RegistrationStatusAccepted,
+	}
+
+	return res, nil
+}
+
+func (cp *CP) StatusNotification(request *core.StatusNotificationRequest) (*core.StatusNotificationConfirmation, error) {
+	cp.log.TRACE.Printf("%T: %+v", request, request)
+
+	if request != nil {
+		cp.mu.Lock()
+		defer cp.mu.Unlock()
+
+		if cp.status == nil {
+			cp.status = request
+			cp.initialized.Broadcast()
+		} else if (request.Timestamp.Equal(cp.status.Timestamp.Time) || request.Timestamp.After(cp.status.Timestamp.Time)) && request.Timestamp.After(time.Now().Add(-30*time.Second)) {
+			cp.status = request
+		} else {
+			cp.log.TRACE.Printf("ignoring status: %s < %s", request.Timestamp.Time, cp.status.Timestamp)
+		}
+	}
+
+	return new(core.StatusNotificationConfirmation), nil
+}
+
+func (cp *CP) DataTransfer(request *core.DataTransferRequest) (*core.DataTransferConfirmation, error) {
+	cp.log.TRACE.Printf("%T: %+v", request, request)
+
+	res := &core.DataTransferConfirmation{
+		Status: core.DataTransferStatusRejected,
+	}
+
+	return res, nil
+}
+
+func (cp *CP) update() {
+	cp.mu.Lock()
+	cp.updated = time.Now()
+	cp.mu.Unlock()
+}
+
+func (cp *CP) Heartbeat(request *core.HeartbeatRequest) (*core.HeartbeatConfirmation, error) {
+	cp.log.TRACE.Printf("%T: %+v", request, request)
+
+	cp.update()
+	res := &core.HeartbeatConfirmation{
+		CurrentTime: types.NewDateTime(time.Now()),
+	}
+
+	if !cp.meterTrickerRunning && cp.meterSupported {
+		Instance().TriggerMeterValueRequest(cp)
+	}
+
+	return res, nil
+}
+
+func (cp *CP) MeterValues(request *core.MeterValuesRequest) (*core.MeterValuesConfirmation, error) {
+	cp.log.TRACE.Printf("%T: %+v", request, request)
+	if request.TransactionId != nil {
+		cp.log.TRACE.Printf("TransactionId: %+v", *request.TransactionId)
+	}
+
+	if request != nil {
+		cp.mu.Lock()
+		cp.setMeterValues(request)
+
+		if energy, ok := cp.measureands[string(types.MeasurandEnergyActiveImportRegister)]; ok {
+			v, _ := strconv.ParseInt(energy.Value, 10, 64)
+			cp.currentTransaction.Charged = v - cp.currentTransaction.MeterValueStart
+		}
+
+		cp.mu.Unlock()
+	}
+
+	return new(core.MeterValuesConfirmation), nil
+}
+
+func getSampleKey(s types.SampledValue) string {
+	if s.Phase != "" {
+		return string(s.Measurand) + "@" + string(s.Phase)
+	}
+
+	return string(s.Measurand)
+}
+
+func (cp *CP) setMeterValues(request *core.MeterValuesRequest) {
+	for _, meterValue := range request.MeterValue {
+		// ignore old meter value requests
+		if meterValue.Timestamp.Time.After(cp.latestMeterValueTimestamp) {
+			for _, sample := range meterValue.SampledValue {
+				cp.measureands[getSampleKey(sample)] = sample
+			}
+		}
+	}
+}
+
+func (cp *CP) StartTransaction(request *core.StartTransactionRequest) (*core.StartTransactionConfirmation, error) {
+	cp.log.TRACE.Printf("%T: %+v", request, request)
+
+	res := &core.StartTransactionConfirmation{
+		IdTagInfo: &types.IdTagInfo{
+			Status: types.AuthorizationStatusAccepted, // accept
+		},
+	}
+
+	// create new transaction
+	if request != nil {
+		if request.Timestamp.After(time.Now().Add(-1 * time.Hour)) { // only respect transactions in the last hour
+			cp.mu.Lock()
+			cp.currentTransaction = NewTransaction(cp.currentTransaction.ID+1, request.IdTag, request.Timestamp.Time, request.MeterStart)
+
+			cp.mu.Unlock()
+
+			res.TransactionId = cp.currentTransaction.ID
+
+			if request.Timestamp.After(time.Now().Add(-30*time.Second)) && cp.meterSupported && !cp.meterTrickerRunning {
+				go func() {
+					cp.log.TRACE.Printf("starting meter value ticker")
+					cp.meterTrickerRunning = true
+					cp.measureDoneCh = make(chan struct{})
+					ticker := time.NewTicker(15 * time.Second)
+
+					defer cp.log.TRACE.Printf("exiting meter value ticker")
+					for {
+						select {
+						case <-ticker.C:
+							Instance().TriggerMeterValueRequest(cp)
+						case <-cp.measureDoneCh:
+							cp.log.TRACE.Printf("returning from meter value requests")
+							cp.meterTrickerRunning = false
+							return
+						}
+					}
+				}()
+			}
+		} else {
+			// TODO: Handle old transactions e.g. store them
+			res.TransactionId = 1 // change 1 to the last known global transaction. Needs persistence
+		}
+	}
+
+	return res, nil
+}
+
+func (cp *CP) StopTransaction(request *core.StopTransactionRequest) (*core.StopTransactionConfirmation, error) {
+	cp.log.TRACE.Printf("%T: %+v", request, request)
+
+	// reset transaction
+	if request != nil {
+		cp.mu.Lock()
+
+		cp.currentTransaction.Finish(request.IdTag, request.Timestamp.Time, request.MeterStop)
+
+		cp.mu.Unlock()
+
+		// TODO: Handle old transaction. Store them, check for the starting transaction event
+	}
+
+	res := &core.StopTransactionConfirmation{
+		IdTagInfo: &types.IdTagInfo{
+			Status: types.AuthorizationStatusAccepted, // accept
+		},
+	}
+
+	if cp.meterSupported {
+		if cp.meterTrickerRunning {
+			cp.measureDoneCh <- struct{}{}
+		}
+
+		Instance().TriggerMeterValueRequest(cp)
+	}
+
+	return res, nil
+}
+
+func (cp *CP) DiagnosticStatusNotification(request *firmware.DiagnosticsStatusNotificationRequest) (*firmware.DiagnosticsStatusNotificationConfirmation, error) {
+	cp.log.TRACE.Printf("%T: %+v", request, request)
+
+	return &firmware.DiagnosticsStatusNotificationConfirmation{}, nil
+}
+
+func (cp *CP) FirmwareStatusNotification(request *firmware.FirmwareStatusNotificationRequest) (*firmware.FirmwareStatusNotificationConfirmation, error) {
+	cp.log.TRACE.Printf("%T: %+v", request, request)
+
+	return &firmware.FirmwareStatusNotificationConfirmation{}, nil
+}
diff --git a/charger/ocpp/cp_transaction.go b/charger/ocpp/cp_transaction.go
new file mode 100644
index 00000000..29b5bf64
--- /dev/null
+++ b/charger/ocpp/cp_transaction.go
@@ -0,0 +1,28 @@
+package ocpp
+
+import "time"
+
+type Transaction struct {
+	ID              int       `yaml:"id"`
+	IDTag           string    `yaml:"idTag"`
+	Start           time.Time `yaml:"startTimestamp"`
+	End             time.Time `yaml:"endTimestamp,omitempty"`
+	MeterValueStart int64     `yaml:"meterValueStart"`
+	MeterValueStop  int64     `yaml:"meterValueStop"`
+	Charged         int64     `yaml:"charged"`
+}
+
+func NewTransaction(id int, idTag string, start time.Time, meterValue int) Transaction {
+	return Transaction{
+		ID:              id,
+		IDTag:           idTag,
+		Start:           start,
+		MeterValueStart: int64(meterValue),
+	}
+}
+
+func (t *Transaction) Finish(idTag string, end time.Time, meterValue int) {
+	t.IDTag = idTag
+	t.MeterValueStop = int64(meterValue)
+	t.End = end
+}
diff --git a/charger/ocpp/cs.go b/charger/ocpp/cs.go
new file mode 100644
index 00000000..453cac65
--- /dev/null
+++ b/charger/ocpp/cs.go
@@ -0,0 +1,95 @@
+package ocpp
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/evcc-io/evcc/util"
+	ocpp16 "github.com/lorenzodonini/ocpp-go/ocpp1.6"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/types"
+)
+
+type CS struct {
+	mu  sync.Mutex
+	log *util.Logger
+	cs  ocpp16.CentralSystem
+	cps map[string]*CP
+}
+
+func (cs *CS) Register(id string, meterSupported bool) *CP {
+	cp := &CP{
+		id:                        id,
+		log:                       util.NewLogger("ocpp-cp"),
+		latestMeterValueTimestamp: time.Now(),
+		measureands:               make(map[string]types.SampledValue),
+		meterSupported:            meterSupported,
+	}
+
+	cp.initialized = sync.NewCond(&cp.mu)
+
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+
+	cs.cps[id] = cp
+
+	return cp
+}
+
+// errorHandler logs error channel
+func (cs *CS) errorHandler(errC <-chan error) {
+	for err := range errC {
+		cs.log.ERROR.Println(err)
+	}
+}
+
+func (cs *CS) chargepointByID(id string) (*CP, error) {
+	cp, ok := cs.cps[id]
+	if !ok {
+		return nil, fmt.Errorf("unknown charge point: %s", id)
+	}
+	return cp, nil
+}
+
+func (cs *CS) NewChargePoint(chargePoint ocpp16.ChargePointConnection) {
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+
+	if _, err := cs.chargepointByID(chargePoint.ID()); err != nil {
+		cs.log.ERROR.Println(err)
+	}
+}
+
+func (cs *CS) ChargePointDisconnected(chargePoint ocpp16.ChargePointConnection) {
+	if _, err := cs.chargepointByID(chargePoint.ID()); err != nil {
+		cs.log.ERROR.Println(err)
+	}
+}
+
+func (cs *CS) CS() ocpp16.CentralSystem {
+	return cs.cs
+}
+
+func (cs *CS) Debug(args ...interface{}) {
+	cs.log.TRACE.Println(args...)
+}
+
+func (cs *CS) Debugf(fmt string, args ...interface{}) {
+	cs.log.TRACE.Printf(fmt, args...)
+}
+
+func (cs *CS) Info(args ...interface{}) {
+	cs.log.DEBUG.Println(args...)
+}
+
+func (cs *CS) Infof(fmt string, args ...interface{}) {
+	cs.log.DEBUG.Printf(fmt, args...)
+}
+
+func (cs *CS) Error(args ...interface{}) {
+	cs.log.ERROR.Println(args...)
+}
+
+func (cs *CS) Errorf(fmt string, args ...interface{}) {
+	cs.log.ERROR.Printf(fmt, args...)
+}
diff --git a/charger/ocpp/cs_core.go b/charger/ocpp/cs_core.go
new file mode 100644
index 00000000..f1c2ee22
--- /dev/null
+++ b/charger/ocpp/cs_core.go
@@ -0,0 +1,131 @@
+package ocpp
+
+import (
+	core "github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/firmware"
+	"github.com/lorenzodonini/ocpp-go/ocpp1.6/remotetrigger"
+)
+
+func (cs *CS) OnAuthorize(chargePointId string, request *core.AuthorizeRequest) (*core.AuthorizeConfirmation, error) {
+	cp, err := cs.chargepointByID(chargePointId)
+	if err != nil {
+		return nil, err
+	}
+
+	return cp.Authorize(request)
+}
+
+func (cs *CS) OnBootNotification(chargePointId string, request *core.BootNotificationRequest) (*core.BootNotificationConfirmation, error) {
+	cp, err := cs.chargepointByID(chargePointId)
+	if err != nil {
+		return nil, err
+	}
+
+	return cp.BootNotification(request)
+}
+
+func (cs *CS) OnDataTransfer(chargePointId string, request *core.DataTransferRequest) (*core.DataTransferConfirmation, error) {
+	cp, err := cs.chargepointByID(chargePointId)
+	if err != nil {
+		return nil, err
+	}
+
+	return cp.DataTransfer(request)
+}
+
+func (cs *CS) OnHeartbeat(chargePointId string, request *core.HeartbeatRequest) (*core.HeartbeatConfirmation, error) {
+	cp, err := cs.chargepointByID(chargePointId)
+	if err != nil {
+		return nil, err
+	}
+
+	return cp.Heartbeat(request)
+}
+
+func (cs *CS) TriggerMeterValueRequest(cp *CP) {
+	callback := func(request *remotetrigger.TriggerMessageConfirmation, err error) {
+		cs.log.TRACE.Printf("TriggerMessageRequest %T: %+v", request, request)
+	}
+
+	if err := cs.cs.TriggerMessage(cp.id, callback, core.MeterValuesFeatureName); err != nil {
+		cs.log.DEBUG.Printf("failed sending TriggerMessageRequest: %s", err)
+	}
+}
+
+func (cs *CS) OnMeterValues(chargePointId string, request *core.MeterValuesRequest) (*core.MeterValuesConfirmation, error) {
+	cp, err := cs.chargepointByID(chargePointId)
+	if err != nil {
+		return nil, err
+	}
+
+	conf, err := cp.MeterValues(request)
+	// if request != nil {
+	// 	cs.log.DEBUG.Printf("%s < %s", request.MeterValue[0].Timestamp, time.Now().Add(-9*time.Second))
+	// 	if request.MeterValue[0].Timestamp.Before(time.Now().Add(-9 * time.Second)) {
+
+	// 		go func() {
+	// 			cs.log.DEBUG.Printf("send extra meter value request to catch up")
+	// 			time.Sleep(3 * time.Second)
+	// 			cs.TriggerMeterValueRequest(cp)
+	// 		}()
+	// 	}
+	// }
+
+	return conf, err
+}
+
+func (cs *CS) OnStatusNotification(chargePointId string, request *core.StatusNotificationRequest) (*core.StatusNotificationConfirmation, error) {
+	cp, err := cs.chargepointByID(chargePointId)
+	if err != nil {
+		return nil, err
+	}
+
+	return cp.StatusNotification(request)
+}
+
+func (cs *CS) OnStartTransaction(chargePointId string, request *core.StartTransactionRequest) (*core.StartTransactionConfirmation, error) {
+	cp, err := cs.chargepointByID(chargePointId)
+	if err != nil {
+		return nil, err
+	}
+
+	return cp.StartTransaction(request)
+}
+
+func (cs *CS) OnStopTransaction(chargePointId string, request *core.StopTransactionRequest) (*core.StopTransactionConfirmation, error) {
+	cp, err := cs.chargepointByID(chargePointId)
+	if err != nil {
+		return nil, err
+	}
+
+	return cp.StopTransaction(request)
+}
+
+func (cs *CS) OnDiagnosticsStatusNotification(chargePointId string, request *firmware.DiagnosticsStatusNotificationRequest) (confirmation *firmware.DiagnosticsStatusNotificationConfirmation, err error) {
+	cp, err := cs.chargepointByID(chargePointId)
+	if err != nil {
+		return nil, err
+	}
+
+	return cp.DiagnosticStatusNotification(request)
+}
+
+func (cs *CS) OnFirmwareStatusNotification(chargePointId string, request *firmware.FirmwareStatusNotificationRequest) (confirmation *firmware.FirmwareStatusNotificationConfirmation, err error) {
+	cp, err := cs.chargepointByID(chargePointId)
+	if err != nil {
+		return nil, err
+	}
+
+	return cp.FirmwareStatusNotification(request)
+
+}
+
+func (cs *CS) TriggerResetRequest(cp *CP, resetType core.ResetType) {
+	callback := func(request *core.ResetConfirmation, err error) {
+		cs.log.TRACE.Printf("TriggerResetRequest %T: %+v", request, request)
+	}
+
+	if err := cs.cs.Reset(cp.id, callback, resetType); err != nil {
+		cs.log.DEBUG.Printf("failed sending TriggerResetRequest: %s", err)
+	}
+}
diff --git a/charger/ocpp/instance.go b/charger/ocpp/instance.go
new file mode 100644
index 00000000..f1af4292
--- /dev/null
+++ b/charger/ocpp/instance.go
@@ -0,0 +1,37 @@
+package ocpp
+
+import (
+	"time"
+
+	"github.com/evcc-io/evcc/util"
+	ocpp16 "github.com/lorenzodonini/ocpp-go/ocpp1.6"
+	"github.com/lorenzodonini/ocpp-go/ocppj"
+)
+
+var instance *CS
+
+func Instance() *CS {
+	if instance == nil {
+		cs := ocpp16.NewCentralSystem(nil, nil)
+
+		instance = &CS{
+			log: util.NewLogger("ocpp"),
+			cps: make(map[string]*CP),
+			cs:  cs,
+		}
+
+		ocppj.SetLogger(instance)
+
+		cs.SetCoreHandler(instance)
+		cs.SetNewChargePointHandler(instance.NewChargePoint)
+		cs.SetChargePointDisconnectedHandler(instance.ChargePointDisconnected)
+		cs.SetFirmwareManagementHandler(instance)
+
+		go Instance().errorHandler(cs.Errors())
+		go cs.Start(8887, "/{ws}")
+
+		time.Sleep(time.Second)
+	}
+
+	return instance
+}
diff --git a/charger/ocpp_decorators.go b/charger/ocpp_decorators.go
new file mode 100644
index 00000000..e1633c20
--- /dev/null
+++ b/charger/ocpp_decorators.go
@@ -0,0 +1,137 @@
+package charger
+
+// Code generated by github.com/evcc-io/evcc/cmd/tools/decorate.go. DO NOT EDIT.
+
+import (
+	"github.com/evcc-io/evcc/api"
+)
+
+func decorateOCPP(base *OCPP, meter func() (float64, error), meterCurrent func() (float64, float64, float64, error), chargeRater func() (float64, error)) api.Charger {
+	switch {
+	case chargeRater == nil && meter == nil && meterCurrent == nil:
+		return base
+
+	case chargeRater == nil && meter != nil && meterCurrent == nil:
+		return &struct {
+			*OCPP
+			api.Meter
+		}{
+			OCPP: base,
+			Meter: &decorateOCPPMeterImpl{
+				meter: meter,
+			},
+		}
+
+	case chargeRater == nil && meter == nil && meterCurrent != nil:
+		return &struct {
+			*OCPP
+			api.MeterCurrent
+		}{
+			OCPP: base,
+			MeterCurrent: &decorateOCPPMeterCurrentImpl{
+				meterCurrent: meterCurrent,
+			},
+		}
+
+	case chargeRater == nil && meter != nil && meterCurrent != nil:
+		return &struct {
+			*OCPP
+			api.Meter
+			api.MeterCurrent
+		}{
+			OCPP: base,
+			Meter: &decorateOCPPMeterImpl{
+				meter: meter,
+			},
+			MeterCurrent: &decorateOCPPMeterCurrentImpl{
+				meterCurrent: meterCurrent,
+			},
+		}
+
+	case chargeRater != nil && meter == nil && meterCurrent == nil:
+		return &struct {
+			*OCPP
+			api.ChargeRater
+		}{
+			OCPP: base,
+			ChargeRater: &decorateOCPPChargeRaterImpl{
+				chargeRater: chargeRater,
+			},
+		}
+
+	case chargeRater != nil && meter != nil && meterCurrent == nil:
+		return &struct {
+			*OCPP
+			api.ChargeRater
+			api.Meter
+		}{
+			OCPP: base,
+			ChargeRater: &decorateOCPPChargeRaterImpl{
+				chargeRater: chargeRater,
+			},
+			Meter: &decorateOCPPMeterImpl{
+				meter: meter,
+			},
+		}
+
+	case chargeRater != nil && meter == nil && meterCurrent != nil:
+		return &struct {
+			*OCPP
+			api.ChargeRater
+			api.MeterCurrent
+		}{
+			OCPP: base,
+			ChargeRater: &decorateOCPPChargeRaterImpl{
+				chargeRater: chargeRater,
+			},
+			MeterCurrent: &decorateOCPPMeterCurrentImpl{
+				meterCurrent: meterCurrent,
+			},
+		}
+
+	case chargeRater != nil && meter != nil && meterCurrent != nil:
+		return &struct {
+			*OCPP
+			api.ChargeRater
+			api.Meter
+			api.MeterCurrent
+		}{
+			OCPP: base,
+			ChargeRater: &decorateOCPPChargeRaterImpl{
+				chargeRater: chargeRater,
+			},
+			Meter: &decorateOCPPMeterImpl{
+				meter: meter,
+			},
+			MeterCurrent: &decorateOCPPMeterCurrentImpl{
+				meterCurrent: meterCurrent,
+			},
+		}
+	}
+
+	return nil
+}
+
+type decorateOCPPChargeRaterImpl struct {
+	chargeRater func() (float64, error)
+}
+
+func (impl *decorateOCPPChargeRaterImpl) ChargedEnergy() (float64, error) {
+	return impl.chargeRater()
+}
+
+type decorateOCPPMeterImpl struct {
+	meter func() (float64, error)
+}
+
+func (impl *decorateOCPPMeterImpl) CurrentPower() (float64, error) {
+	return impl.meter()
+}
+
+type decorateOCPPMeterCurrentImpl struct {
+	meterCurrent func() (float64, float64, float64, error)
+}
+
+func (impl *decorateOCPPMeterCurrentImpl) Currents() (float64, float64, float64, error) {
+	return impl.meterCurrent()
+}
diff --git a/provider/mqtt.go b/provider/mqtt.go
index 1fdfe801..b5786c71 100644
--- a/provider/mqtt.go
+++ b/provider/mqtt.go
@@ -107,7 +107,7 @@ func (m *Mqtt) newReceiver() *msgHandler {
 	h := &msgHandler{
 		topic:    m.topic,
 		scale:    m.scale,
-		mux:      util.NewWaiter(m.timeout, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }),
+		wait:     util.NewWaiter(m.timeout, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }),
 		pipeline: m.pipeline,
 	}
 
diff --git a/provider/mqtt_handler.go b/provider/mqtt_handler.go
index 4da97389..af44237c 100644
--- a/provider/mqtt_handler.go
+++ b/provider/mqtt_handler.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/evcc-io/evcc/provider/pipeline"
@@ -11,7 +12,8 @@ import (
 )
 
 type msgHandler struct {
-	mux      *util.Waiter
+	mux      sync.Mutex
+	wait     *util.Waiter
 	scale    float64
 	topic    string
 	pipeline *pipeline.Pipeline
@@ -23,18 +25,18 @@ func (h *msgHandler) receive(payload string) {
 	defer h.mux.Unlock()
 
 	h.payload = payload
-	h.mux.Update()
+	h.wait.Update()
 }
 
 // hasValue returned the received and processed payload as string
 func (h *msgHandler) hasValue() (string, error) {
-	h.mux.Lock()
-	defer h.mux.Unlock()
-
-	if late := h.mux.Overdue(); late > 0 {
+	if late := h.wait.Overdue(); late > 0 {
 		return "", fmt.Errorf("%s outdated: %v", h.topic, late.Truncate(time.Second))
 	}
 
+	h.mux.Lock()
+	defer h.mux.Unlock()
+
 	if h.pipeline != nil {
 		b, err := h.pipeline.Process([]byte(h.payload))
 		return string(b), err
diff --git a/provider/sma/device.go b/provider/sma/device.go
index ae429880..d6644bbc 100644
--- a/provider/sma/device.go
+++ b/provider/sma/device.go
@@ -15,7 +15,8 @@ type Device struct {
 	*sunny.Device
 
 	log    *util.Logger
-	mux    *util.Waiter
+	mux    sync.Mutex
+	wait   *util.Waiter
 	values map[sunny.ValueID]interface{}
 	once   sync.Once
 }
@@ -40,7 +41,7 @@ func (d *Device) UpdateValues() error {
 	values, err := d.Device.GetValues()
 	if err == nil {
 		err = mergo.Merge(&d.values, values, mergo.WithOverride)
-		d.mux.Update()
+		d.wait.Update()
 	}
 
 	return err
@@ -53,7 +54,7 @@ func (d *Device) Values() (map[sunny.ValueID]interface{}, error) {
 	d.mux.Lock()
 	defer d.mux.Unlock()
 
-	if late := d.mux.Overdue(); late > 0 {
+	if late := d.wait.Overdue(); late > 0 {
 		return nil, fmt.Errorf("update timeout: %v", late.Truncate(time.Second))
 	}
 
diff --git a/provider/sma/discover.go b/provider/sma/discover.go
index fa55c35d..f891bbad 100644
--- a/provider/sma/discover.go
+++ b/provider/sma/discover.go
@@ -65,7 +65,7 @@ func (d *Discoverer) createDevice(device *sunny.Device) *Device {
 	return &Device{
 		Device: device,
 		log:    d.log,
-		mux:    util.NewWaiter(udpTimeout, func() { d.log.DEBUG.Println("wait for initial value") }),
+		wait:   util.NewWaiter(udpTimeout, func() { d.log.DEBUG.Println("wait for initial value") }),
 		values: make(map[sunny.ValueID]interface{}),
 	}
 }
diff --git a/provider/socket.go b/provider/socket.go
index 73e776da..4669d4bd 100644
--- a/provider/socket.go
+++ b/provider/socket.go
@@ -5,6 +5,7 @@ import (
 	"math"
 	"net/http"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/evcc-io/evcc/util"
@@ -21,7 +22,8 @@ const retryDelay = 5 * time.Second
 type Socket struct {
 	*request.Helper
 	log     *util.Logger
-	mux     *util.Waiter
+	mux     sync.Mutex
+	wait    *util.Waiter
 	url     string
 	headers map[string]string
 	scale   float64
@@ -62,7 +64,7 @@ func NewSocketProviderFromConfig(other map[string]interface{}) (IntProvider, err
 	p := &Socket{
 		log:     log,
 		Helper:  request.NewHelper(log),
-		mux:     util.NewWaiter(cc.Timeout, func() { log.DEBUG.Println("wait for initial value") }),
+		wait:    util.NewWaiter(cc.Timeout, func() { log.DEBUG.Println("wait for initial value") }),
 		url:     url,
 		headers: cc.Headers,
 		scale:   cc.Scale,
@@ -129,11 +131,11 @@ func (p *Socket) listen() {
 				v, err := jq.Query(p.jq, b)
 				if err == nil {
 					p.val = v
-					p.mux.Update()
+					p.wait.Update()
 				}
 			} else {
 				p.val = string(b)
-				p.mux.Update()
+				p.wait.Update()
 			}
 			p.mux.Unlock()
 		}
@@ -141,13 +143,13 @@ func (p *Socket) listen() {
 }
 
 func (p *Socket) hasValue() (interface{}, error) {
-	p.mux.Lock()
-	defer p.mux.Unlock()
-
-	if late := p.mux.Overdue(); late > 0 {
+	if late := p.wait.Overdue(); late > 0 {
 		return nil, fmt.Errorf("outdated: %v", late.Truncate(time.Second))
 	}
 
+	p.mux.Lock()
+	defer p.mux.Unlock()
+
 	return p.val, nil
 }
 
diff --git a/templates/definition/meter/solax-hybrid-cloud.yaml b/templates/definition/meter/solax-hybrid-cloud.yaml
index 85df99a1..e1e52211 100644
--- a/templates/definition/meter/solax-hybrid-cloud.yaml
+++ b/templates/definition/meter/solax-hybrid-cloud.yaml
@@ -38,7 +38,7 @@ render: |
   power:
   {{- if eq .usage "grid" }}
     source: http
-    uri: https://www.eu.solaxcloud.com:9443/proxy/api/getRealtimeInfo.do?tokenId=={{ .tokenid }}&sn={{ .serial}}
+    uri: https://www.eu.solaxcloud.com:9443/proxy/api/getRealtimeInfo.do?tokenId={{ .tokenid }}&sn={{ .serial}}
     jq: .result.feedinpower
     cache: 2m30s
     scale: -1
diff --git a/util/waiter.go b/util/waiter.go
index d842716a..a6c0efce 100644
--- a/util/waiter.go
+++ b/util/waiter.go
@@ -9,63 +9,54 @@ var waitInitialTimeout = 10 * time.Second
 
 // Waiter provides monitoring of receive timeouts and reception of initial value
 type Waiter struct {
-	sync.Mutex
+	mu      sync.Mutex
 	log     func()
-	cond    *sync.Cond
 	updated time.Time
 	timeout time.Duration
+	initial chan bool
 }
 
 // NewWaiter creates new waiter
 func NewWaiter(timeout time.Duration, logInitialWait func()) *Waiter {
-	p := &Waiter{
+	return &Waiter{
 		log:     logInitialWait,
 		timeout: timeout,
+		initial: make(chan bool),
 	}
-	p.cond = sync.NewCond(p)
-	return p
 }
 
 // Update is called when client has received data. Update resets the timeout counter.
-// Waiter MUST be locked when calling Update.
 func (p *Waiter) Update() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	p.updated = time.Now()
-	p.cond.Broadcast()
+
+	select {
+	case <-p.initial:
+	default:
+		close(p.initial)
+	}
 }
 
 // Overdue waits for initial update and returns the duration since the last update
-// in excess of timeout. Waiter MUST be locked when calling Overdue.
+// in excess of timeout.
 func (p *Waiter) Overdue() time.Duration {
-	if p.updated.IsZero() {
+	select {
+	case <-p.initial:
+	default:
 		p.log()
 
-		c := make(chan struct{})
-
-		go func() {
-			defer close(c)
-			p.Lock() // establish lock once go routine has started
-			for p.updated.IsZero() {
-				p.cond.Wait()
-			}
-		}()
-
-		// release lock so external updates can occur
-		p.Unlock()
-
 		select {
-		case <-c:
-			// initial value received, lock established
+		case <-p.initial:
 		case <-time.After(waitInitialTimeout):
-			// establish lock per contract of `Update()``
-			p.Lock()
-			p.Update() // unblock the sync.Cond
-			p.Unlock()
-			<-c                     // wait for goroutine, re-establish lock
-			p.updated = time.Time{} // reset to "initial value missing"
 			return waitInitialTimeout
 		}
 	}
 
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	if elapsed := time.Since(p.updated); p.timeout != 0 && elapsed > p.timeout {
 		return elapsed
 	}


Test patch:
diff --git a/util/waiter_test.go b/util/waiter_test.go
index 7ff903b9..2c6cdf4b 100644
--- a/util/waiter_test.go
+++ b/util/waiter_test.go
@@ -19,16 +19,12 @@ func TestWaiterInitialUpdateInTime(t *testing.T) {
 
 		go func() {
 			time.Sleep(testTimeout / 2)
-			w.Lock()
 			w.Update()
-			w.Unlock()
 		}()
 
-		w.Lock()
 		if elapsed := w.Overdue(); elapsed != 0 {
 			t.Errorf("expected %v, got %v", 0, elapsed)
 		}
-		w.Unlock()
 	}
 }
 
@@ -36,30 +32,21 @@ func TestWaiterInitialUpdateNotReceived(t *testing.T) {
 	for _, timeout := range []time.Duration{0, testTimeout} {
 		w := NewWaiter(timeout, func() {})
 
-		w.Lock()
 		if elapsed := w.Overdue(); elapsed != waitInitialTimeout {
 			t.Errorf("expected %v, got %v", waitInitialTimeout, elapsed)
 		}
-		w.Unlock()
 	}
 }
 
 func TestWaiterUpdateInTime(t *testing.T) {
 	w := NewWaiter(testTimeout, func() {})
-	w.Lock()
 	w.Update()
-	w.Unlock()
 
 	go func() {
 		time.Sleep(testTimeout / 2)
-		w.Lock()
 		w.Update()
-		w.Unlock()
 	}()
 
-	w.Lock()
-	defer w.Unlock()
-
 	if elapsed := w.Overdue(); elapsed != 0 {
 		t.Errorf("expected %v, got %v", 0, elapsed)
 	}
@@ -67,15 +54,10 @@ func TestWaiterUpdateInTime(t *testing.T) {
 
 func TestWaiterUpdateNotReceived(t *testing.T) {
 	w := NewWaiter(testTimeout, func() {})
-	w.Lock()
 	w.Update()
-	w.Unlock()
 
 	time.Sleep(2 * testTimeout)
 
-	w.Lock()
-	defer w.Unlock()
-
 	if elapsed := w.Overdue(); elapsed < 2*testTimeout {
 		t.Errorf("expected >%v, got %v", 2*testTimeout, elapsed)
 	}