From bd810bd89bb26c4f686a2ceb31c38d34009a8c8e Mon Sep 17 00:00:00 2001 From: andig Date: Thu, 10 Dec 2020 12:00:13 +0100 Subject: [PATCH] Refactor plugin configuration (#511) --- charger/config_test.go | 1 + charger/openwb.go | 45 +++-- cmd/config.go | 9 +- cmd/setup.go | 10 +- meter/config_test.go | 1 + provider/config.go | 357 ++++++++++++-------------------------- provider/helper.go | 18 ++ provider/http.go | 106 ++++++----- provider/javascript.go | 110 +++++++----- provider/modbus.go | 22 ++- provider/mqtt.go | 233 ++++++++----------------- provider/mqtt/client.go | 136 +++++++++++++++ provider/mqtt/registry.go | 40 +++++ provider/script.go | 148 +++++++++------- provider/socket.go | 99 ++++++----- server/mqtt.go | 6 +- vehicle/config_test.go | 1 + 17 files changed, 724 insertions(+), 618 deletions(-) create mode 100644 provider/helper.go create mode 100644 provider/mqtt/client.go create mode 100644 provider/mqtt/registry.go diff --git a/charger/config_test.go b/charger/config_test.go index 46ae6aace..5994b6892 100644 --- a/charger/config_test.go +++ b/charger/config_test.go @@ -9,6 +9,7 @@ import ( func TestChargers(t *testing.T) { acceptable := []string{ "invalid plugin type: ...", + "missing mqtt broker configuration", "mqtt not configured", "invalid charger type: nrgkick-bluetooth", "NRGKick bluetooth is only supported on linux", diff --git a/charger/openwb.go b/charger/openwb.go index 1797aebd2..d90d680f0 100644 --- a/charger/openwb.go +++ b/charger/openwb.go @@ -8,6 +8,7 @@ import ( "github.com/andig/evcc/charger/openwb" "github.com/andig/evcc/meter" "github.com/andig/evcc/provider" + "github.com/andig/evcc/provider/mqtt" "github.com/andig/evcc/util" ) @@ -24,11 +25,10 @@ type OpenWB struct { // NewOpenWBFromConfig creates a new configurable charger func NewOpenWBFromConfig(other map[string]interface{}) (api.Charger, error) { cc := struct { - Broker string - User, Password string - Topic string - ID int - Timeout time.Duration + mqtt.Config `mapstructure:",squash"` + Topic string + ID int + Timeout time.Duration }{ Topic: "openWB", ID: 1, @@ -41,37 +41,50 @@ func NewOpenWBFromConfig(other map[string]interface{}) (api.Charger, error) { log := util.NewLogger("openwb") - clientID := provider.MqttClientID() - client, err := provider.NewMqttClient(log, cc.Broker, cc.User, cc.Password, clientID, 1) + clientID := mqtt.ClientID() + client, err := mqtt.RegisteredClient(log, cc.Broker, cc.User, cc.Password, clientID, 1) if err != nil { return nil, err } + // getters + boolG := func(topic string) func() (bool, error) { + return provider.NewMqtt(log, client, topic, "", 1, cc.Timeout).BoolGetter() + } + + floatG := func(topic string) func() (float64, error) { + return provider.NewMqtt(log, client, topic, "", 1, cc.Timeout).FloatGetter() + } + // check if loadpoint configured - configured := client.BoolGetter(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.ConfiguredTopic), cc.Timeout) + configured := boolG(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.ConfiguredTopic)) if isConfigured, err := configured(); err != nil || !isConfigured { return nil, fmt.Errorf("openWB loadpoint %d is not configured", cc.ID) } // adapt plugged/charging to status - plugged := client.BoolGetter(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.PluggedTopic), cc.Timeout) - charging := client.BoolGetter(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.ChargingTopic), cc.Timeout) + plugged := boolG(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.PluggedTopic)) + charging := boolG(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.ChargingTopic)) status := provider.NewOpenWBStatusProvider(plugged, charging).StringGetter // remaining getters - enabled := client.BoolGetter(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.EnabledTopic), cc.Timeout) + enabled := boolG(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.EnabledTopic)) // setters - enable := client.BoolSetter("enable", fmt.Sprintf("%s/set/lp%d/%s", cc.Topic, cc.ID, openwb.EnabledTopic), "") - maxcurrent := client.IntSetter("maxcurrent", fmt.Sprintf("%s/set/lp%d/%s", cc.Topic, cc.ID, openwb.MaxCurrentTopic), "") + enable := provider.NewMqtt(log, client, + fmt.Sprintf("%s/set/lp%d/%s", cc.Topic, cc.ID, openwb.EnabledTopic), + "", 1, cc.Timeout).BoolSetter("enable") + maxcurrent := provider.NewMqtt(log, client, + fmt.Sprintf("%s/set/lp%d/%s", cc.Topic, cc.ID, openwb.MaxCurrentTopic), + "", 1, cc.Timeout).IntSetter("maxcurrent") // meter getters - power := client.FloatGetter(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.ChargePowerTopic), 1, cc.Timeout) - totalEnergy := client.FloatGetter(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.ChargeTotalEnergyTopic), 1, cc.Timeout) + power := floatG(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.ChargePowerTopic)) + totalEnergy := floatG(fmt.Sprintf("%s/lp/%d/%s", cc.Topic, cc.ID, openwb.ChargeTotalEnergyTopic)) var currents []func() (float64, error) for i := 1; i <= 3; i++ { - current := client.FloatGetter(fmt.Sprintf("%s/lp/%d/%s%d", cc.Topic, cc.ID, openwb.CurrentTopic, i), 1, cc.Timeout) + current := floatG(fmt.Sprintf("%s/lp/%d/%s%d", cc.Topic, cc.ID, openwb.CurrentTopic, i)) currents = append(currents, current) } diff --git a/cmd/config.go b/cmd/config.go index 0e038f93d..5a5383d2f 100644 --- a/cmd/config.go +++ b/cmd/config.go @@ -7,7 +7,7 @@ import ( "github.com/andig/evcc/api" "github.com/andig/evcc/charger" "github.com/andig/evcc/meter" - "github.com/andig/evcc/provider" + "github.com/andig/evcc/provider/mqtt" "github.com/andig/evcc/push" "github.com/andig/evcc/server" "github.com/andig/evcc/vehicle" @@ -20,7 +20,7 @@ type config struct { Profile bool Levels map[string]string Interval time.Duration - Mqtt provider.MqttConfig + Mqtt mqttConfig Influx server.InfluxConfig HEMS typedConfig Messaging messagingConfig @@ -31,6 +31,11 @@ type config struct { LoadPoints []map[string]interface{} } +type mqttConfig struct { + mqtt.Config `mapstructure:",squash"` + Topic string +} + type qualifiedConfig struct { Name, Type string Other map[string]interface{} `mapstructure:",remain"` diff --git a/cmd/setup.go b/cmd/setup.go index b6dca2007..9060fbbfd 100644 --- a/cmd/setup.go +++ b/cmd/setup.go @@ -9,7 +9,7 @@ import ( "github.com/andig/evcc/core" "github.com/andig/evcc/hems" - "github.com/andig/evcc/provider" + "github.com/andig/evcc/provider/mqtt" "github.com/andig/evcc/push" "github.com/andig/evcc/server" "github.com/andig/evcc/util" @@ -46,14 +46,14 @@ func configureDatabase(conf server.InfluxConfig, loadPoints []core.LoadPointAPI, } // setup mqtt -func configureMQTT(conf provider.MqttConfig) { +func configureMQTT(conf mqttConfig) { log := util.NewLogger("mqtt") - clientID := provider.MqttClientID() + clientID := mqtt.ClientID() var err error - provider.MQTT, err = provider.NewMqttClient(log, conf.Broker, conf.User, conf.Password, clientID, 1) + mqtt.Instance, err = mqtt.RegisteredClient(log, conf.Broker, conf.User, conf.Password, clientID, 1) if err != nil { - log.FATAL.Fatalf("failed configuring hems: %v", err) + log.FATAL.Fatalf("failed configuring mqtt: %v", err) } } diff --git a/meter/config_test.go b/meter/config_test.go index 1b4015b1a..41d5eda5a 100644 --- a/meter/config_test.go +++ b/meter/config_test.go @@ -9,6 +9,7 @@ import ( func TestMeters(t *testing.T) { acceptable := []string{ "invalid plugin type: ...", + "missing mqtt broker configuration", "mqtt not configured", "not a SunSpec device", "connect: no route to host", diff --git a/provider/config.go b/provider/config.go index d7fbc5739..8b10ba48d 100644 --- a/provider/config.go +++ b/provider/config.go @@ -1,161 +1,97 @@ package provider import ( - "errors" "fmt" "strings" - "time" - - "github.com/andig/evcc/util" ) -const ( - execTimeout = 5 * time.Second +// provider types +type ( + IntProvider interface { + IntGetter() func() (int64, error) + } + StringProvider interface { + StringGetter() func() (string, error) + } + FloatProvider interface { + FloatGetter() func() (float64, error) + } + BoolProvider interface { + BoolGetter() func() (bool, error) + } + SetIntProvider interface { + IntSetter(param string) func(int64) error + } + SetBoolProvider interface { + BoolSetter(param string) func(bool) error + } ) +type providerRegistry map[string]func(map[string]interface{}) (IntProvider, error) + +func (r providerRegistry) Add(name string, factory func(map[string]interface{}) (IntProvider, error)) { + if _, exists := r[name]; exists { + panic(fmt.Sprintf("cannot register duplicate plugin type: %s", name)) + } + r[name] = factory +} + +func (r providerRegistry) Get(name string) (func(map[string]interface{}) (IntProvider, error), error) { + factory, exists := r[name] + if !exists { + return nil, fmt.Errorf("invalid plugin type: %s", name) + } + return factory, nil +} + +var registry providerRegistry = make(map[string]func(map[string]interface{}) (IntProvider, error)) + // Config is the general provider config type Config struct { Type string Other map[string]interface{} `mapstructure:",remain"` } -// mqttConfig is the specific mqtt getter/setter configuration -type mqttConfig struct { - Topic, Payload string // Payload only applies to setters - Scale float64 - Timeout time.Duration -} +// NewIntGetterFromConfig creates a IntGetter from config +func NewIntGetterFromConfig(config Config) (res func() (int64, error), err error) { + factory, err := registry.Get(strings.ToLower(config.Type)) + if err == nil { + var provider IntProvider + provider, err = factory(config.Other) -// scriptConfig is the specific script getter/setter configuration -type scriptConfig struct { - Cmd string - Timeout time.Duration - Cache time.Duration -} - -// MQTT singleton -var MQTT *MqttClient - -func mqttFromConfig(other map[string]interface{}) (mqttConfig, error) { - pc := mqttConfig{Scale: 1} - if err := util.DecodeOther(other, &pc); err != nil { - return pc, err + if err == nil { + res = provider.IntGetter() + } } - if MQTT == nil { - return pc, errors.New("mqtt not configured") - } - - return pc, nil -} - -func scriptFromConfig(other map[string]interface{}) (scriptConfig, error) { - var pc scriptConfig - if err := util.DecodeOther(other, &pc); err != nil { - return pc, err - } - - if pc.Timeout == 0 { - pc.Timeout = execTimeout - } - - return pc, nil -} - -// NewFloatGetterFromConfig creates a FloatGetter from config -func NewFloatGetterFromConfig(config Config) (res func() (float64, error), err error) { - switch strings.ToLower(config.Type) { - case "calc": - res, err = NewCalcFromConfig(config.Other) - case "http": - var prov *HTTP - if prov, err = NewHTTPProviderFromConfig(config.Other); err == nil { - res = prov.FloatGetter - } - case "js": - var prov *Javascript - if prov, err = NewJavascriptProviderFromConfig(config.Other); err == nil { - res = prov.FloatGetter - } - case "websocket", "ws": - var prov *Socket - if prov, err = NewSocketProviderFromConfig(config.Other); err == nil { - res = prov.FloatGetter - } - case "mqtt": - if pc, err := mqttFromConfig(config.Other); err == nil { - res = MQTT.FloatGetter(pc.Topic, pc.Scale, pc.Timeout) - } - case "script": - var pc scriptConfig - if pc, err = scriptFromConfig(config.Other); err != nil { - break - } - - var prov *Script - if prov, err = NewScriptProvider(pc.Timeout); err == nil { - res = prov.FloatGetter(pc.Cmd) - } - - if pc.Cache > 0 { - res = NewCached(res, pc.Cache).FloatGetter() - } - case "modbus": - var prov *Modbus - if prov, err = NewModbusFromConfig(config.Other); err == nil { - res = prov.FloatGetter - } - default: - return nil, fmt.Errorf("invalid plugin type: %s", config.Type) + if err == nil && res == nil { + err = fmt.Errorf("invalid plugin type: %s", config.Type) } return } -// NewIntGetterFromConfig creates a IntGetter from config -func NewIntGetterFromConfig(config Config) (res func() (int64, error), err error) { - switch strings.ToLower(config.Type) { - case "http": - var prov *HTTP - if prov, err = NewHTTPProviderFromConfig(config.Other); err == nil { - res = prov.IntGetter - } - case "js": - var prov *Javascript - if prov, err = NewJavascriptProviderFromConfig(config.Other); err == nil { - res = prov.IntGetter - } - case "websocket", "ws": - var prov *Socket - if prov, err = NewSocketProviderFromConfig(config.Other); err == nil { - res = prov.IntGetter - } - case "mqtt": - var pc mqttConfig - if pc, err = mqttFromConfig(config.Other); err == nil { - res = MQTT.IntGetter(pc.Topic, int64(pc.Scale), pc.Timeout) - } - case "script": - var pc scriptConfig - if pc, err = scriptFromConfig(config.Other); err != nil { - break - } +// NewFloatGetterFromConfig creates a FloatGetter from config +func NewFloatGetterFromConfig(config Config) (res func() (float64, error), err error) { + switch typ := strings.ToLower(config.Type); typ { + case "calc": + res, err = NewCalcFromConfig(config.Other) - var prov *Script - if prov, err = NewScriptProvider(pc.Timeout); err == nil { - res = prov.IntGetter(pc.Cmd) - } - - if pc.Cache > 0 { - res = NewCached(res, pc.Cache).IntGetter() - } - case "modbus": - var prov *Modbus - if prov, err = NewModbusFromConfig(config.Other); err == nil { - res = prov.IntGetter - } default: - err = fmt.Errorf("invalid plugin type: %s", config.Type) + var factory func(map[string]interface{}) (IntProvider, error) + factory, err = registry.Get(typ) + if err == nil { + var provider IntProvider + provider, err = factory(config.Other) + + if prov, ok := provider.(FloatProvider); ok { + res = prov.FloatGetter() + } + } + + if err == nil && res == nil { + err = fmt.Errorf("invalid plugin type: %s", config.Type) + } } return @@ -163,45 +99,25 @@ func NewIntGetterFromConfig(config Config) (res func() (int64, error), err error // NewStringGetterFromConfig creates a StringGetter from config func NewStringGetterFromConfig(config Config) (res func() (string, error), err error) { - switch strings.ToLower(config.Type) { - case "http": - var prov *HTTP - if prov, err = NewHTTPProviderFromConfig(config.Other); err == nil { - res = prov.StringGetter - } - case "js": - var prov *Javascript - if prov, err = NewJavascriptProviderFromConfig(config.Other); err == nil { - res = prov.StringGetter - } - case "websocket", "ws": - var prov *Socket - if prov, err = NewSocketProviderFromConfig(config.Other); err == nil { - res = prov.StringGetter - } - case "mqtt": - var pc mqttConfig - if pc, err = mqttFromConfig(config.Other); err == nil { - res = MQTT.StringGetter(pc.Topic, pc.Timeout) - } - case "script": - var pc scriptConfig - if pc, err = scriptFromConfig(config.Other); err != nil { - break - } - - var prov *Script - if prov, err = NewScriptProvider(pc.Timeout); err == nil { - res = prov.StringGetter(pc.Cmd) - } - - if pc.Cache > 0 { - res = NewCached(res, pc.Cache).StringGetter() - } + switch typ := strings.ToLower(config.Type); typ { case "combined", "openwb": res, err = NewOpenWBStatusProviderFromConfig(config.Other) + default: - err = fmt.Errorf("invalid plugin type: %s", config.Type) + var factory func(map[string]interface{}) (IntProvider, error) + factory, err = registry.Get(typ) + if err == nil { + var provider IntProvider + provider, err = factory(config.Other) + + if prov, ok := provider.(StringProvider); ok { + res = prov.StringGetter() + } + } + + if err == nil && res == nil { + err = fmt.Errorf("invalid plugin type: %s", config.Type) + } } return @@ -209,42 +125,17 @@ func NewStringGetterFromConfig(config Config) (res func() (string, error), err e // NewBoolGetterFromConfig creates a BoolGetter from config func NewBoolGetterFromConfig(config Config) (res func() (bool, error), err error) { - switch strings.ToLower(config.Type) { - case "http": - var prov *HTTP - if prov, err = NewHTTPProviderFromConfig(config.Other); err == nil { - res = prov.BoolGetter - } - case "js": - var prov *Javascript - if prov, err = NewJavascriptProviderFromConfig(config.Other); err == nil { - res = prov.BoolGetter - } - case "websocket", "ws": - var prov *Socket - if prov, err = NewSocketProviderFromConfig(config.Other); err == nil { - res = prov.BoolGetter - } - case "mqtt": - var pc mqttConfig - if pc, err = mqttFromConfig(config.Other); err == nil { - res = MQTT.BoolGetter(pc.Topic, pc.Timeout) - } - case "script": - var pc scriptConfig - if pc, err = scriptFromConfig(config.Other); err != nil { - break - } + factory, err := registry.Get(strings.ToLower(config.Type)) + if err == nil { + var provider IntProvider + provider, err = factory(config.Other) - var prov *Script - if prov, err = NewScriptProvider(pc.Timeout); err == nil { - res = prov.BoolGetter(pc.Cmd) + if prov, ok := provider.(BoolProvider); ok { + res = prov.BoolGetter() } + } - if pc.Cache > 0 { - res = NewCached(res, pc.Cache).BoolGetter() - } - default: + if err == nil && res == nil { err = fmt.Errorf("invalid plugin type: %s", config.Type) } @@ -253,29 +144,18 @@ func NewBoolGetterFromConfig(config Config) (res func() (bool, error), err error // NewIntSetterFromConfig creates a IntSetter from config func NewIntSetterFromConfig(param string, config Config) (res func(int64) error, err error) { - switch strings.ToLower(config.Type) { - case "http": - var prov *HTTP - if prov, err = NewHTTPProviderFromConfig(config.Other); err == nil { - res = prov.IntSetter - } - case "mqtt": - var pc mqttConfig - if pc, err = mqttFromConfig(config.Other); err == nil { - res = MQTT.IntSetter(param, pc.Topic, pc.Payload) - } - case "script": - var pc scriptConfig - if pc, err = scriptFromConfig(config.Other); err != nil { - break - } + factory, err := registry.Get(strings.ToLower(config.Type)) + if err == nil { + var provider IntProvider + provider, err = factory(config.Other) - var prov *Script - if prov, err = NewScriptProvider(pc.Timeout); err == nil { - res = prov.IntSetter(param, pc.Cmd) + if prov, ok := provider.(SetIntProvider); ok { + res = prov.IntSetter(param) } - default: - err = fmt.Errorf("invalid setter type %s", config.Type) + } + + if err == nil && res == nil { + err = fmt.Errorf("invalid plugin type: %s", config.Type) } return @@ -283,29 +163,18 @@ func NewIntSetterFromConfig(param string, config Config) (res func(int64) error, // NewBoolSetterFromConfig creates a BoolSetter from config func NewBoolSetterFromConfig(param string, config Config) (res func(bool) error, err error) { - switch strings.ToLower(config.Type) { - case "http": - var prov *HTTP - if prov, err = NewHTTPProviderFromConfig(config.Other); err == nil { - res = prov.BoolSetter - } - case "mqtt": - var pc mqttConfig - if pc, err = mqttFromConfig(config.Other); err == nil { - res = MQTT.BoolSetter(param, pc.Topic, pc.Payload) - } - case "script": - var pc scriptConfig - if pc, err = scriptFromConfig(config.Other); err != nil { - break - } + factory, err := registry.Get(strings.ToLower(config.Type)) + if err == nil { + var provider IntProvider + provider, err = factory(config.Other) - var prov *Script - if prov, err = NewScriptProvider(pc.Timeout); err == nil { - res = prov.BoolSetter(param, pc.Cmd) + if prov, ok := provider.(SetBoolProvider); ok { + res = prov.BoolSetter(param) } - default: - err = fmt.Errorf("invalid setter type %s", config.Type) + } + + if err == nil && res == nil { + err = fmt.Errorf("invalid plugin type: %s", config.Type) } return diff --git a/provider/helper.go b/provider/helper.go new file mode 100644 index 000000000..ab70f22d1 --- /dev/null +++ b/provider/helper.go @@ -0,0 +1,18 @@ +package provider + +import ( + "fmt" + + "github.com/andig/evcc/util" +) + +// setFormattedValue formats a message template of returns the value formatted as %v is template is empty +func setFormattedValue(message, param string, v interface{}) (string, error) { + if message == "" { + return fmt.Sprintf("%v", v), nil + } + + return util.ReplaceFormatted(message, map[string]interface{}{ + param: v, + }) +} diff --git a/provider/http.go b/provider/http.go index 173138174..c864fd8f6 100644 --- a/provider/http.go +++ b/provider/http.go @@ -25,6 +25,10 @@ type HTTP struct { jq *gojq.Query } +func init() { + registry.Add("http", NewHTTPProviderFromConfig) +} + // Auth is the authorization config type Auth struct { Type, User, Password string @@ -42,7 +46,7 @@ func NewAuth(log *util.Logger, auth Auth, headers map[string]string) error { } // NewHTTPProviderFromConfig creates a HTTP provider -func NewHTTPProviderFromConfig(other map[string]interface{}) (*HTTP, error) { +func NewHTTPProviderFromConfig(other map[string]interface{}) (IntProvider, error) { cc := struct { URI, Method string Headers map[string]string @@ -111,64 +115,88 @@ func (p *HTTP) request(body ...string) ([]byte, error) { } // FloatGetter parses float from request -func (p *HTTP) FloatGetter() (float64, error) { - s, err := p.StringGetter() - if err != nil { - return 0, err - } +func (p *HTTP) FloatGetter() func() (float64, error) { + g := p.StringGetter() - f, err := strconv.ParseFloat(s, 64) - if err == nil && p.scale != 0 { - f *= p.scale - } + return func() (float64, error) { + s, err := g() + if err != nil { + return 0, err + } - return f, err + f, err := strconv.ParseFloat(s, 64) + if err == nil && p.scale != 0 { + f *= p.scale + } + + return f, err + } } // IntGetter parses int64 from request -func (p *HTTP) IntGetter() (int64, error) { - f, err := p.FloatGetter() - return int64(math.Round(f)), err +func (p *HTTP) IntGetter() func() (int64, error) { + g := p.FloatGetter() + + return func() (int64, error) { + f, err := g() + return int64(math.Round(f)), err + } } // StringGetter sends string request -func (p *HTTP) StringGetter() (string, error) { - b, err := p.request() - if err != nil { +func (p *HTTP) StringGetter() func() (string, error) { + return func() (string, error) { + b, err := p.request() + if err != nil { + return string(b), err + } + + if p.jq != nil { + v, err := jq.Query(p.jq, b) + return fmt.Sprintf("%v", v), err + } + return string(b), err } - - if p.jq != nil { - v, err := jq.Query(p.jq, b) - return fmt.Sprintf("%v", v), err - } - - return string(b), err } // BoolGetter parses bool from request -func (p *HTTP) BoolGetter() (bool, error) { - s, err := p.StringGetter() - return util.Truish(s), err +func (p *HTTP) BoolGetter() func() (bool, error) { + g := p.StringGetter() + + return func() (bool, error) { + s, err := g() + return util.Truish(s), err + } +} + +func (p *HTTP) set(param string, val interface{}) error { + body, err := setFormattedValue(p.body, param, val) + + if err == nil { + _, err = p.request(body) + } + + return err } // IntSetter sends int request -func (p *HTTP) IntSetter(param int64) error { - body := util.FormatValue(p.body, param) - _, err := p.request(body) - return err +func (p *HTTP) IntSetter(param string) func(int64) error { + return func(val int64) error { + return p.set(param, val) + } } // StringSetter sends string request -func (p *HTTP) StringSetter(param string) error { - body := util.FormatValue(p.body, param) - _, err := p.request(body) - return err +func (p *HTTP) StringSetter(param string) func(string) error { + return func(val string) error { + return p.set(param, val) + } } // BoolSetter sends bool request -func (p *HTTP) BoolSetter(param bool) error { - body := util.FormatValue(p.body, param) - _, err := p.request(body) - return err +func (p *HTTP) BoolSetter(param string) func(bool) error { + return func(val bool) error { + return p.set(param, val) + } } diff --git a/provider/javascript.go b/provider/javascript.go index 4f61242ea..b877733a8 100644 --- a/provider/javascript.go +++ b/provider/javascript.go @@ -12,8 +12,12 @@ type Javascript struct { script string } +func init() { + registry.Add("js", NewJavascriptProviderFromConfig) +} + // NewJavascriptProviderFromConfig creates a HTTP provider -func NewJavascriptProviderFromConfig(other map[string]interface{}) (*Javascript, error) { +func NewJavascriptProviderFromConfig(other map[string]interface{}) (IntProvider, error) { cc := struct { Script string }{} @@ -34,62 +38,82 @@ func NewJavascriptProviderFromConfig(other map[string]interface{}) (*Javascript, } // FloatGetter parses float from request -func (p *Javascript) FloatGetter() (res float64, err error) { - v, err := p.vm.Eval(p.script) - if err == nil { - res, err = v.ToFloat() - } +func (p *Javascript) FloatGetter() func() (float64, error) { + return func() (res float64, err error) { + v, err := p.vm.Eval(p.script) + if err == nil { + res, err = v.ToFloat() + } - return res, err + return res, err + } } // IntGetter parses int64 from request -func (p *Javascript) IntGetter() (res int64, err error) { - v, err := p.vm.Eval(p.script) - if err == nil { - res, err = v.ToInteger() - } +func (p *Javascript) IntGetter() func() (int64, error) { + return func() (res int64, err error) { + v, err := p.vm.Eval(p.script) + if err == nil { + res, err = v.ToInteger() + } - return res, err + return res, err + } } // StringGetter sends string request -func (p *Javascript) StringGetter() (res string, err error) { - v, err := p.vm.Eval(p.script) - if err == nil { - res, err = v.ToString() - } +func (p *Javascript) StringGetter() func() (string, error) { + return func() (res string, err error) { + v, err := p.vm.Eval(p.script) + if err == nil { + res, err = v.ToString() + } - return res, err + return res, err + } } // BoolGetter parses bool from request -func (p *Javascript) BoolGetter() (res bool, err error) { - v, err := p.vm.Eval(p.script) - if err == nil { - res, err = v.ToBoolean() - } +func (p *Javascript) BoolGetter() func() (bool, error) { + return func() (res bool, err error) { + v, err := p.vm.Eval(p.script) + if err == nil { + res, err = v.ToBoolean() + } - return res, err + return res, err + } } -// // IntSetter sends int request -// func (p *Javascript) IntSetter(param int64) error { -// body := util.FormatValue(p.body, param) -// _, err := p.request(body) -// return err -// } +// IntSetter sends int request +func (p *Javascript) IntSetter(param string) func(int64) error { + return func(val int64) error { + err := p.vm.Set(param, val) + if err == nil { + _, err = p.vm.Eval(p.script) + } + return err + } +} -// // StringSetter sends string request -// func (p *Javascript) StringSetter(param string) error { -// body := util.FormatValue(p.body, param) -// _, err := p.request(body) -// return err -// } +// StringSetter sends string request +func (p *Javascript) StringSetter(param string) func(string) error { + return func(val string) error { + err := p.vm.Set(param, val) + if err == nil { + _, err = p.vm.Eval(p.script) + } + return err + } +} -// // BoolSetter sends bool request -// func (p *Javascript) BoolSetter(param bool) error { -// body := util.FormatValue(p.body, param) -// _, err := p.request(body) -// return err -// } +// BoolSetter sends bool request +func (p *Javascript) BoolSetter(param string) func(bool) error { + return func(val bool) error { + err := p.vm.Set(param, val) + if err == nil { + _, err = p.vm.Eval(p.script) + } + return err + } +} diff --git a/provider/modbus.go b/provider/modbus.go index 6d323e1ed..dc4567ebb 100644 --- a/provider/modbus.go +++ b/provider/modbus.go @@ -21,8 +21,12 @@ type Modbus struct { scale float64 } +func init() { + registry.Add("modbus", NewModbusFromConfig) +} + // NewModbusFromConfig creates Modbus plugin -func NewModbusFromConfig(other map[string]interface{}) (*Modbus, error) { +func NewModbusFromConfig(other map[string]interface{}) (IntProvider, error) { cc := struct { Model string modbus.Settings `mapstructure:",squash"` @@ -112,7 +116,11 @@ func NewModbusFromConfig(other map[string]interface{}) (*Modbus, error) { } // FloatGetter executes configured modbus read operation and implements func() (float64, error) -func (m *Modbus) FloatGetter() (float64, error) { +func (m *Modbus) FloatGetter() func() (float64, error) { + return m.floatGetter +} + +func (m *Modbus) floatGetter() (float64, error) { var res meters.MeasurementResult var err error @@ -168,7 +176,11 @@ func (m *Modbus) FloatGetter() (float64, error) { } // IntGetter executes configured modbus read operation and implements provider.IntGetter -func (m *Modbus) IntGetter() (int64, error) { - res, err := m.FloatGetter() - return int64(math.Round(res)), err +func (m *Modbus) IntGetter() func() (int64, error) { + g := m.FloatGetter() + + return func() (int64, error) { + res, err := g() + return int64(math.Round(res)), err + } } diff --git a/provider/mqtt.go b/provider/mqtt.go index 9205ca5b8..c9cba4812 100644 --- a/provider/mqtt.go +++ b/provider/mqtt.go @@ -1,231 +1,152 @@ package provider import ( + "errors" "fmt" "math" - "math/rand" "strconv" - "sync" "time" + "github.com/andig/evcc/provider/mqtt" "github.com/andig/evcc/util" - mqtt "github.com/eclipse/paho.mqtt.golang" ) -const ( - connectTimeout = 2 * time.Second - publishTimeout = 2 * time.Second -) - -// MqttClientID created unique mqtt client id -func MqttClientID() string { - pid := rand.Int31() - return fmt.Sprintf("evcc-%d", pid) +// Mqtt provider +type Mqtt struct { + log *util.Logger + client *mqtt.Client + topic string + payload string + scale float64 + timeout time.Duration } -// MqttConfig is the public configuration -type MqttConfig struct { - Broker string - User string - Password string - Topic string +func init() { + registry.Add("mqtt", NewMqttFromConfig) } -// MqttClient is a paho publisher -type MqttClient struct { - log *util.Logger - mux sync.Mutex - Client mqtt.Client - broker string - Qos byte - listener map[string]func(string) -} - -// NewMqttClient creates new publisher for paho -func NewMqttClient( - log *util.Logger, - broker string, - user string, - password string, - clientID string, - qos byte, -) (*MqttClient, error) { - broker = util.DefaultPort(broker, 1883) - log.INFO.Printf("connecting %s at %s", clientID, broker) - - mc := &MqttClient{ - log: log, - broker: broker, - Qos: qos, - listener: make(map[string]func(string)), +// NewMqttFromConfig creates Mqtt provider +func NewMqttFromConfig(other map[string]interface{}) (IntProvider, error) { + cc := struct { + mqtt.Config `mapstructure:",squash"` + Topic, Payload string // Payload only applies to setters + Scale float64 + Timeout time.Duration + }{ + Scale: 1, } - options := mqtt.NewClientOptions() - options.AddBroker(broker) - options.SetUsername(user) - options.SetPassword(password) - options.SetClientID(clientID) - options.SetCleanSession(true) - options.SetAutoReconnect(true) - options.SetOnConnectHandler(mc.ConnectionHandler) - options.SetConnectionLostHandler(mc.ConnectionLostHandler) - options.SetConnectTimeout(connectTimeout) - - client := mqtt.NewClient(options) - if token := client.Connect(); token.Wait() && token.Error() != nil { - return nil, fmt.Errorf("error connecting: %w", token.Error()) + if err := util.DecodeOther(other, &cc); err != nil { + return nil, err } - mc.Client = client - return mc, nil -} + log := util.NewLogger("mqtt") -// ConnectionLostHandler logs cause of connection loss as warning -func (m *MqttClient) ConnectionLostHandler(client mqtt.Client, reason error) { - m.log.ERROR.Printf("%s connection lost: %v", m.broker, reason.Error()) -} + var err error + client := mqtt.Instance -// ConnectionHandler restores listeners -func (m *MqttClient) ConnectionHandler(client mqtt.Client) { - m.log.DEBUG.Printf("%s connected", m.broker) - - m.mux.Lock() - defer m.mux.Unlock() - - for topic, l := range m.listener { - m.log.TRACE.Printf("%s subscribe %s", m.broker, topic) - go m.listen(topic, l) + if client == nil && cc.Config.Broker == "" { + return nil, errors.New("missing mqtt broker configuration") } -} -// Listen validates uniqueness and registers and attaches listener -func (m *MqttClient) Listen(topic string, callback func(string)) { - m.mux.Lock() - if _, ok := m.listener[topic]; ok { - m.log.FATAL.Fatalf("%s: duplicate listener not allowed", topic) - } - m.listener[topic] = callback - m.mux.Unlock() - - m.listen(topic, callback) -} - -// listen attaches listener to topic -func (m *MqttClient) listen(topic string, callback func(string)) { - token := m.Client.Subscribe(topic, m.Qos, func(c mqtt.Client, msg mqtt.Message) { - s := string(msg.Payload()) - if len(s) > 0 { - callback(s) + if client == nil { + client, err = mqtt.RegisteredClient(log, cc.Broker, cc.User, cc.Password, mqtt.ClientID(), 1) + if err != nil { + return nil, err } - }) - m.WaitForToken(token) + } + + m := NewMqtt(log, client, cc.Topic, cc.Payload, cc.Scale, cc.Timeout) + + return m, err +} + +// NewMqtt creates mqtt provider for given topic +func NewMqtt(log *util.Logger, client *mqtt.Client, topic string, payload string, scale float64, timeout time.Duration) *Mqtt { + m := &Mqtt{ + log: log, + client: client, + topic: topic, + payload: payload, + scale: scale, + timeout: timeout, + } + return m } // FloatGetter creates handler for float64 from MQTT topic that returns cached value -func (m *MqttClient) FloatGetter(topic string, scale float64, timeout time.Duration) func() (float64, error) { +func (m *Mqtt) FloatGetter() func() (float64, error) { h := &msgHandler{ log: m.log, - mux: util.NewWaiter(timeout, func() { m.log.TRACE.Printf("%s wait for initial value", topic) }), - topic: topic, - scale: scale, + topic: m.topic, + scale: m.scale, + mux: util.NewWaiter(m.timeout, func() { m.log.TRACE.Printf("%s wait for initial value", m.topic) }), } - m.Listen(topic, h.Receive) + m.client.Listen(m.topic, h.receive) return h.floatGetter } // IntGetter creates handler for int64 from MQTT topic that returns cached value -func (m *MqttClient) IntGetter(topic string, scale int64, timeout time.Duration) func() (int64, error) { +func (m *Mqtt) IntGetter() func() (int64, error) { h := &msgHandler{ log: m.log, - mux: util.NewWaiter(timeout, func() { m.log.TRACE.Printf("%s wait for initial value", topic) }), - topic: topic, - scale: float64(scale), + topic: m.topic, + scale: float64(m.scale), + mux: util.NewWaiter(m.timeout, func() { m.log.TRACE.Printf("%s wait for initial value", m.topic) }), } - m.Listen(topic, h.Receive) + m.client.Listen(m.topic, h.receive) return h.intGetter } // StringGetter creates handler for string from MQTT topic that returns cached value -func (m *MqttClient) StringGetter(topic string, timeout time.Duration) func() (string, error) { +func (m *Mqtt) StringGetter() func() (string, error) { h := &msgHandler{ log: m.log, - mux: util.NewWaiter(timeout, func() { m.log.TRACE.Printf("%s wait for initial value", topic) }), - topic: topic, + topic: m.topic, + mux: util.NewWaiter(m.timeout, func() { m.log.TRACE.Printf("%s wait for initial value", m.topic) }), } - m.Listen(topic, h.Receive) + m.client.Listen(m.topic, h.receive) return h.stringGetter } // BoolGetter creates handler for string from MQTT topic that returns cached value -func (m *MqttClient) BoolGetter(topic string, timeout time.Duration) func() (bool, error) { +func (m *Mqtt) BoolGetter() func() (bool, error) { h := &msgHandler{ log: m.log, - mux: util.NewWaiter(timeout, func() { m.log.TRACE.Printf("%s wait for initial value", topic) }), - topic: topic, + topic: m.topic, + mux: util.NewWaiter(m.timeout, func() { m.log.TRACE.Printf("%s wait for initial value", m.topic) }), } - m.Listen(topic, h.Receive) + m.client.Listen(m.topic, h.receive) return h.boolGetter } -// formatValue formats a message template of returns the value formatted as %v is template is empty -func (m *MqttClient) formatValue(param, message string, v interface{}) (string, error) { - if message == "" { - return fmt.Sprintf("%v", v), nil - } - - return util.ReplaceFormatted(message, map[string]interface{}{ - param: v, - }) -} - // IntSetter publishes topic with parameter replaced by int value -func (m *MqttClient) IntSetter(param, topic, message string) func(int64) error { +func (m *Mqtt) IntSetter(param string) func(int64) error { return func(v int64) error { - payload, err := m.formatValue(param, message, v) + payload, err := setFormattedValue(m.payload, param, v) if err != nil { return err } - m.log.TRACE.Printf("send %s: '%s'", topic, payload) - token := m.Client.Publish(topic, m.Qos, false, payload) - if token.WaitTimeout(publishTimeout) { - return token.Error() - } - - return fmt.Errorf("%s send timeout", topic) + m.log.TRACE.Printf("send %s: '%s'", m.topic, payload) + return m.client.Publish(m.topic, false, payload) } } // BoolSetter invokes script with parameter replaced by bool value -func (m *MqttClient) BoolSetter(param, topic, message string) func(bool) error { +func (m *Mqtt) BoolSetter(param string) func(bool) error { return func(v bool) error { - payload, err := m.formatValue(param, message, v) + payload, err := setFormattedValue(m.payload, param, v) if err != nil { return err } - m.log.TRACE.Printf("send %s: '%s'", topic, payload) - token := m.Client.Publish(topic, m.Qos, false, payload) - if token.WaitTimeout(publishTimeout) { - return token.Error() - } - - return fmt.Errorf("%s send timeout", topic) - } -} - -// WaitForToken synchronously waits until token operation completed -func (m *MqttClient) WaitForToken(token mqtt.Token) { - if token.WaitTimeout(publishTimeout) { - if token.Error() != nil { - m.log.ERROR.Printf("error: %s", token.Error()) - } - } else { - m.log.DEBUG.Println("timeout") + m.log.TRACE.Printf("send %s: '%s'", m.topic, payload) + return m.client.Publish(m.topic, false, payload) } } @@ -237,7 +158,7 @@ type msgHandler struct { payload string } -func (h *msgHandler) Receive(payload string) { +func (h *msgHandler) receive(payload string) { h.log.TRACE.Printf("recv %s: '%s'", h.topic, payload) h.mux.Lock() diff --git a/provider/mqtt/client.go b/provider/mqtt/client.go new file mode 100644 index 000000000..cfe8ba449 --- /dev/null +++ b/provider/mqtt/client.go @@ -0,0 +1,136 @@ +package mqtt + +import ( + "fmt" + "math/rand" + "sync" + "time" + + "github.com/andig/evcc/util" + mqtt "github.com/eclipse/paho.mqtt.golang" +) + +const ( + connectTimeout = 2 * time.Second + publishTimeout = 2 * time.Second +) + +// Instance is the paho Mqtt client singleton +var Instance *Client + +// ClientID created unique mqtt client id +func ClientID() string { + pid := rand.Int31() + return fmt.Sprintf("evcc-%d", pid) +} + +// Config is the public configuration +type Config struct { + Broker string + User string + Password string +} + +// Client encapsulates mqtt publish/subscribe functions +type Client struct { + log *util.Logger + mux sync.Mutex + Client mqtt.Client + broker string + Qos byte + listener map[string]func(string) +} + +// NewClient creates new Mqtt publisher +func NewClient(log *util.Logger, broker, user, password, clientID string, qos byte) (*Client, error) { + broker = util.DefaultPort(broker, 1883) + log.INFO.Printf("connecting %s at %s", clientID, broker) + + mc := &Client{ + log: log, + broker: broker, + Qos: qos, + listener: make(map[string]func(string)), + } + + options := mqtt.NewClientOptions() + options.AddBroker(broker) + options.SetUsername(user) + options.SetPassword(password) + options.SetClientID(clientID) + options.SetCleanSession(true) + options.SetAutoReconnect(true) + options.SetOnConnectHandler(mc.ConnectionHandler) + options.SetConnectionLostHandler(mc.ConnectionLostHandler) + options.SetConnectTimeout(connectTimeout) + + client := mqtt.NewClient(options) + if token := client.Connect(); token.Wait() && token.Error() != nil { + return nil, fmt.Errorf("error connecting: %w", token.Error()) + } + + mc.Client = client + + return mc, nil +} + +// ConnectionLostHandler logs cause of connection loss as warning +func (m *Client) ConnectionLostHandler(client mqtt.Client, reason error) { + m.log.ERROR.Printf("%s connection lost: %v", m.broker, reason.Error()) +} + +// ConnectionHandler restores listeners +func (m *Client) ConnectionHandler(client mqtt.Client) { + m.log.DEBUG.Printf("%s connected", m.broker) + + m.mux.Lock() + defer m.mux.Unlock() + + for topic, l := range m.listener { + m.log.TRACE.Printf("%s subscribe %s", m.broker, topic) + go m.listen(topic, l) + } +} + +// Publish synchronously pulishes payload using client qos +func (m *Client) Publish(topic string, retained bool, payload interface{}) error { + token := m.Client.Publish(topic, m.Qos, retained, payload) + if token.WaitTimeout(publishTimeout) { + return token.Error() + } + return nil +} + +// Listen validates uniqueness and registers and attaches listener +func (m *Client) Listen(topic string, callback func(string)) { + m.mux.Lock() + if _, ok := m.listener[topic]; ok { + m.log.FATAL.Fatalf("%s: duplicate listener not allowed", topic) + } + m.listener[topic] = callback + m.mux.Unlock() + + m.listen(topic, callback) +} + +// listen attaches listener to topic +func (m *Client) listen(topic string, callback func(string)) { + token := m.Client.Subscribe(topic, m.Qos, func(c mqtt.Client, msg mqtt.Message) { + s := string(msg.Payload()) + if len(s) > 0 { + callback(s) + } + }) + m.WaitForToken(token) +} + +// WaitForToken synchronously waits until token operation completed +func (m *Client) WaitForToken(token mqtt.Token) { + if token.WaitTimeout(publishTimeout) { + if token.Error() != nil { + m.log.ERROR.Printf("error: %s", token.Error()) + } + } else { + m.log.DEBUG.Println("timeout") + } +} diff --git a/provider/mqtt/registry.go b/provider/mqtt/registry.go new file mode 100644 index 000000000..5a8d50d96 --- /dev/null +++ b/provider/mqtt/registry.go @@ -0,0 +1,40 @@ +package mqtt + +import ( + "fmt" + + "github.com/andig/evcc/util" +) + +type clientRegistry map[string]*Client + +func (r clientRegistry) Add(broker string, client *Client) { + if _, exists := r[broker]; exists { + panic(fmt.Sprintf("cannot register duplicate broker: %s", broker)) + } + r[broker] = client +} + +func (r clientRegistry) Get(broker string) (*Client, error) { + client, exists := r[broker] + if !exists { + return nil, fmt.Errorf("missing mqtt broker configuration: %s", broker) + } + return client, nil +} + +// registry is the Mqtt client registry +var registry clientRegistry = make(map[string]*Client) + +// RegisteredClient reuses an registered Mqtt publisher or creates a new one +func RegisteredClient(log *util.Logger, broker, user, password, clientID string, qos byte) (*Client, error) { + client, err := registry.Get(broker) + + if err != nil { + if client, err = NewClient(log, broker, user, password, ClientID(), qos); err == nil { + registry.Add(broker, client) + } + } + + return client, err +} diff --git a/provider/script.go b/provider/script.go index 54a1a2229..2ca8891b5 100644 --- a/provider/script.go +++ b/provider/script.go @@ -15,61 +15,95 @@ import ( // Script implements shell script-based providers and setters type Script struct { log *util.Logger + script string timeout time.Duration + cache time.Duration + updated time.Time + val string + err error +} + +func init() { + registry.Add("script", NewScriptProviderFromConfig) +} + +// NewScriptProviderFromConfig creates a script provider. +func NewScriptProviderFromConfig(other map[string]interface{}) (IntProvider, error) { + cc := struct { + Cmd string + Timeout time.Duration + Cache time.Duration + }{ + Timeout: 5 * time.Second, + } + + if err := util.DecodeOther(other, &cc); err != nil { + return nil, err + } + + return NewScriptProvider(cc.Cmd, cc.Timeout, cc.Cache) } // NewScriptProvider creates a script provider. // Script execution is aborted after given timeout. -func NewScriptProvider(timeout time.Duration) (*Script, error) { +func NewScriptProvider(script string, timeout time.Duration, cache time.Duration) (*Script, error) { s := &Script{ - log: util.NewLogger("exec"), + log: util.NewLogger("script"), + script: script, timeout: timeout, + cache: cache, } + + return s, nil +} + +func (e *Script) exec(script string) (string, error) { + args, err := shellquote.Split(script) + if err != nil { + return "", err + } + + ctx, cancel := context.WithTimeout(context.Background(), e.timeout) + defer cancel() + + cmd := exec.CommandContext(ctx, args[0], args[1:]...) + b, err := cmd.Output() + + s := strings.TrimSpace(string(b)) + + if err != nil { + // use STDOUT if available + var ee *exec.ExitError + if errors.As(err, &ee) { + s = strings.TrimSpace(string(ee.Stderr)) + } + + e.log.ERROR.Printf("%s: %s", strings.Join(args, " "), s) + return "", err + } + + e.log.TRACE.Printf("%s: %s", strings.Join(args, " "), s) return s, nil } // StringGetter returns string from exec result. Only STDOUT is considered. -func (e *Script) StringGetter(script string) func() (string, error) { - args, err := shellquote.Split(script) - if err != nil { - panic(err) - } else if len(args) < 1 { - panic("exec: missing script") - } - - // return func to access cached value +func (e *Script) StringGetter() func() (string, error) { return func() (string, error) { - ctx, cancel := context.WithTimeout(context.Background(), e.timeout) - defer cancel() - - cmd := exec.CommandContext(ctx, args[0], args[1:]...) - b, err := cmd.Output() - - s := strings.TrimSpace(string(b)) - - if err != nil { - // use STDOUT if available - var ee *exec.ExitError - if errors.As(err, &ee) { - s = strings.TrimSpace(string(ee.Stderr)) - } - - e.log.ERROR.Printf("%s: %s", strings.Join(args, " "), s) - return "", err + if time.Since(e.updated) > e.cache { + e.val, e.err = e.exec(e.script) + e.updated = time.Now() } - e.log.TRACE.Printf("%s: %s", strings.Join(args, " "), s) - return s, nil + return e.val, e.err } } // IntGetter parses int64 from exec result -func (e *Script) IntGetter(script string) func() (int64, error) { - exec := e.StringGetter(script) +func (e *Script) IntGetter() func() (int64, error) { + g := e.StringGetter() - // return func to access cached value return func() (int64, error) { - s, err := exec() + s, err := g() if err != nil { return 0, err } @@ -79,12 +113,11 @@ func (e *Script) IntGetter(script string) func() (int64, error) { } // FloatGetter parses float from exec result -func (e *Script) FloatGetter(script string) func() (float64, error) { - exec := e.StringGetter(script) +func (e *Script) FloatGetter() func() (float64, error) { + g := e.StringGetter() - // return func to access cached value return func() (float64, error) { - s, err := exec() + s, err := g() if err != nil { return 0, err } @@ -94,12 +127,11 @@ func (e *Script) FloatGetter(script string) func() (float64, error) { } // BoolGetter parses bool from exec result. "on", "true" and 1 are considered truish. -func (e *Script) BoolGetter(script string) func() (bool, error) { - exec := e.StringGetter(script) +func (e *Script) BoolGetter() func() (bool, error) { + g := e.StringGetter() - // return func to access cached value return func() (bool, error) { - s, err := exec() + s, err := g() if err != nil { return false, err } @@ -109,41 +141,33 @@ func (e *Script) BoolGetter(script string) func() (bool, error) { } // IntSetter invokes script with parameter replaced by int value -func (e *Script) IntSetter(param, script string) func(int64) error { +func (e *Script) IntSetter(param string) func(int64) error { // return func to access cached value return func(i int64) error { - cmd, err := util.ReplaceFormatted(script, map[string]interface{}{ + cmd, err := util.ReplaceFormatted(e.script, map[string]interface{}{ param: i, }) - if err != nil { - return err + + if err == nil { + _, err = e.exec(cmd) } - exec := e.StringGetter(cmd) - if _, err := exec(); err != nil { - return err - } - - return nil + return err } } // BoolSetter invokes script with parameter replaced by bool value -func (e *Script) BoolSetter(param, script string) func(bool) error { +func (e *Script) BoolSetter(param string) func(bool) error { // return func to access cached value return func(b bool) error { - cmd, err := util.ReplaceFormatted(script, map[string]interface{}{ + cmd, err := util.ReplaceFormatted(e.script, map[string]interface{}{ param: b, }) - if err != nil { - return err + + if err == nil { + _, err = e.exec(cmd) } - exec := e.StringGetter(cmd) - if _, err := exec(); err != nil { - return err - } - - return nil + return err } } diff --git a/provider/socket.go b/provider/socket.go index b835826f6..7b374da14 100644 --- a/provider/socket.go +++ b/provider/socket.go @@ -29,8 +29,13 @@ type Socket struct { val interface{} } +func init() { + registry.Add("ws", NewSocketProviderFromConfig) + registry.Add("websocket", NewSocketProviderFromConfig) +} + // NewSocketProviderFromConfig creates a HTTP provider -func NewSocketProviderFromConfig(other map[string]interface{}) (*Socket, error) { +func NewSocketProviderFromConfig(other map[string]interface{}) (IntProvider, error) { cc := struct { URI string Headers map[string]string @@ -133,66 +138,74 @@ func (p *Socket) hasValue() (interface{}, error) { } // StringGetter sends string request -func (p *Socket) StringGetter() (string, error) { - v, err := p.hasValue() - if err != nil { - return "", err - } +func (p *Socket) StringGetter() func() (string, error) { + return func() (string, error) { + v, err := p.hasValue() + if err != nil { + return "", err + } - return jq.String(v) + return jq.String(v) + } } // FloatGetter parses float from string getter -func (p *Socket) FloatGetter() (float64, error) { - v, err := p.hasValue() - if err != nil { - return 0, err - } - - // v is always string when jq not used - if p.jq == nil { - v, err = strconv.ParseFloat(v.(string), 64) +func (p *Socket) FloatGetter() func() (float64, error) { + return func() (float64, error) { + v, err := p.hasValue() if err != nil { return 0, err } - } - f, err := jq.Float64(v) - return f * p.scale, err + // v is always string when jq not used + if p.jq == nil { + v, err = strconv.ParseFloat(v.(string), 64) + if err != nil { + return 0, err + } + } + + f, err := jq.Float64(v) + return f * p.scale, err + } } // IntGetter parses int64 from float getter -func (p *Socket) IntGetter() (int64, error) { - v, err := p.hasValue() - if err != nil { - return 0, err - } - - // v is always string when jq not used - if p.jq == nil { - v, err = strconv.ParseInt(v.(string), 10, 64) +func (p *Socket) IntGetter() func() (int64, error) { + return func() (int64, error) { + v, err := p.hasValue() if err != nil { return 0, err } + + // v is always string when jq not used + if p.jq == nil { + v, err = strconv.ParseInt(v.(string), 10, 64) + if err != nil { + return 0, err + } + } + + i, err := jq.Int64(v) + f := float64(i) * p.scale + + return int64(math.Round(f)), err } - - i, err := jq.Int64(v) - f := float64(i) * p.scale - - return int64(math.Round(f)), err } // BoolGetter parses bool from string getter -func (p *Socket) BoolGetter() (bool, error) { - v, err := p.hasValue() - if err != nil { - return false, err - } +func (p *Socket) BoolGetter() func() (bool, error) { + return func() (bool, error) { + v, err := p.hasValue() + if err != nil { + return false, err + } - // v is always string when jq not used - if p.jq == nil { - v = util.Truish(v.(string)) - } + // v is always string when jq not used + if p.jq == nil { + v = util.Truish(v.(string)) + } - return jq.Bool(v) + return jq.Bool(v) + } } diff --git a/server/mqtt.go b/server/mqtt.go index f210bcdae..c6b8177b1 100644 --- a/server/mqtt.go +++ b/server/mqtt.go @@ -7,13 +7,13 @@ import ( "github.com/andig/evcc/api" "github.com/andig/evcc/core" - "github.com/andig/evcc/provider" + "github.com/andig/evcc/provider/mqtt" "github.com/andig/evcc/util" ) // MQTT is the MQTT server. It uses the MQTT client for publishing. type MQTT struct { - Handler *provider.MqttClient + Handler *mqtt.Client root string } @@ -24,7 +24,7 @@ func NewMQTT(root string) *MQTT { } return &MQTT{ - Handler: provider.MQTT, + Handler: mqtt.Instance, root: root, } } diff --git a/vehicle/config_test.go b/vehicle/config_test.go index 503aae423..f7968988f 100644 --- a/vehicle/config_test.go +++ b/vehicle/config_test.go @@ -9,6 +9,7 @@ import ( func TestVehicles(t *testing.T) { acceptable := []string{ "invalid plugin type: ...", + "missing mqtt broker configuration", "received status code 404 (INVALID PARAMS)", // Nissan "missing personID", "401 Unauthorized",