Store status in redis and restore on startup

Author: xanecs

broker.go

@@ -8,6 +8,7 @@ import (
 type broker struct {
 	client *redis.Client
 	pubsub *redis.PubSub
+	topic  string
 }
 
 func newBroker(cfg config.RedisConfig) (*broker, error) {
@@ -23,6 +24,7 @@ func newBroker(cfg config.RedisConfig) (*broker, error) {
 	return &broker{
 		client: client,
 		pubsub: pubsub,
+		topic:  cfg.Topic,
 	}, nil
 }
 
@@ -37,6 +39,21 @@ func (b *broker) listen(messages chan string, errors chan error) {
 	}
 }
 
+func (b *broker) updateStatus(status chan Status, errors chan error) {
+	for {
+		s := <-status
+		err := b.client.HMSet(b.topic+":"+s.Device, s.Params).Err()
+		if err != nil {
+			errors <- err
+		}
+
+	}
+}
+
+func (b *broker) fetchStatus(device string) (map[string]string, error) {
+	return b.client.HGetAll(b.topic + ":" + device).Result()
+}
+
 func (b *broker) Close() {
 	b.pubsub.Close()
 	b.client.Close()

driver/direct.go

@@ -2,6 +2,7 @@ package driver
 
 import (
 	"errors"
+	"fmt"
 
 	"github.com/xanecs/lighthouse/config"
 
@@ -54,8 +55,17 @@ func (d *directDriver) HandleMessage(action string, p params) error {
 	return d.write()
 }
 
-func (d *directDriver) Status() map[string]interface{} {
-	return map[string]interface{}{"power": d.state}
+func (d *directDriver) Status() map[string]string {
+	return map[string]string{"power": fmt.Sprint(d.state)}
+}
+
+func (d *directDriver) Restore(status map[string]string) error {
+	v, ok := status["power"]
+	if !ok {
+		return errors.New("Missing parameter power")
+	}
+	d.state = v == trueStr
+	return d.write()
 }
 
 func newDirectDriver(cfg config.DeviceConfig, connection gobot.Connection) (*directDriver, error) {

driver/driver.go

@@ -22,7 +22,8 @@ const (
 // Device represents a device that can handle messages
 type Device interface {
 	HandleMessage(string, params) error
-	Status() map[string]interface{}
+	Status() map[string]string
+	Restore(map[string]string) error
 }
 
 // NewDriver creates a new Driver from a device config

driver/params.go

@@ -2,6 +2,10 @@ package driver
 
 import "errors"
 
+const (
+	trueStr = "true"
+)
+
 type params map[string]interface{}
 
 func (p params) getBool(key string) (bool, error) {

driver/pwm.go

@@ -2,6 +2,8 @@ package driver
 
 import (
 	"errors"
+	"fmt"
+	"strconv"
 
 	"github.com/xanecs/lighthouse/config"
 
@@ -79,8 +81,26 @@ func (p *pwmDriver) HandleMessage(action string, par params) error {
 	return p.write()
 }
 
-func (p *pwmDriver) Status() map[string]interface{} {
-	return map[string]interface{}{"power": p.power, "brightness": p.brightness / 255.0}
+func (p *pwmDriver) Status() map[string]string {
+	return map[string]string{"power": fmt.Sprint(p.power), "brightness": fmt.Sprint(float64(p.brightness) / 255.0)}
+}
+
+func (p *pwmDriver) Restore(status map[string]string) error {
+	pwr, ok := status["power"]
+	if !ok {
+		return errors.New("Missing parameter 'power'")
+	}
+	p.power = pwr == trueStr
+	brightStr, ok := status["brightness"]
+	if !ok {
+		return errors.New("Missing parameter 'brightness'")
+	}
+	bright, err := strconv.ParseFloat(brightStr, 64)
+	if err != nil {
+		return err
+	}
+	p.brightness = byte(bright * 255)
+	return p.write()
 }
 
 func newPwmDriver(cfg config.DeviceConfig, connection gobot.Connection) (*pwmDriver, error) {

driver/rgb.go

@@ -2,6 +2,7 @@ package driver
 
 import (
 	"errors"
+	"fmt"
 
 	"github.com/xanecs/lighthouse/config"
 
@@ -77,8 +78,31 @@ func (r *rgbDriver) HandleMessage(action string, p params) error {
 	return r.write()
 }
 
-func (r *rgbDriver) Status() map[string]interface{} {
-	return map[string]interface{}{"power": r.power, "color": r.color}
+func (r *rgbDriver) Status() map[string]string {
+	return map[string]string{
+		"power": fmt.Sprint(r.power),
+		"red":   fmt.Sprint(r.color.red),
+		"green": fmt.Sprint(r.color.green),
+		"blue":  fmt.Sprint(r.color.blue),
+	}
+}
+
+func (r *rgbDriver) Restore(status map[string]string) error {
+	par := make(params)
+	for k, v := range status {
+		par[k] = v
+	}
+	clr, err := parseColor(par)
+	if err != nil {
+		return err
+	}
+	r.color = clr
+	powerStr, ok := status["power"]
+	if !ok {
+		return errors.New("Missing parameter 'power'")
+	}
+	r.power = powerStr == trueStr
+	return r.write()
 }
 
 func (r *rgbDriver) Driver() gobot.Device {

driver/servo.go

@@ -2,6 +2,8 @@ package driver
 
 import (
 	"errors"
+	"fmt"
+	"strconv"
 
 	"github.com/xanecs/lighthouse/config"
 	"gobot.io/x/gobot"
@@ -72,8 +74,27 @@ func (s *servoDriver) HandleMessage(action string, p params) error {
 	return s.write()
 }
 
-func (s *servoDriver) Status() map[string]interface{} {
-	return map[string]interface{}{"power": s.power, "angle": s.angle}
+func (s *servoDriver) Status() map[string]string {
+	return map[string]string{"power": fmt.Sprint(s.power), "angle": fmt.Sprint(s.angle)}
+}
+
+func (s *servoDriver) Restore(status map[string]string) error {
+	powerStr, ok := status["power"]
+	if !ok {
+		return errors.New("Missing parameter 'power'")
+	}
+	s.power = powerStr == trueStr
+
+	angleStr, ok := status["angle"]
+	if !ok {
+		return errors.New("Missing parameter 'angle'")
+	}
+	angle, err := strconv.ParseUint(angleStr, 10, 8)
+	if err != nil {
+		return err
+	}
+	s.angle = uint8(angle)
+	return s.write()
 }
 
 func newServoDriver(cfg config.DeviceConfig, connection gobot.Connection) (*servoDriver, error) {

hardware.go

@@ -2,6 +2,8 @@ package main
 
 import (
 	"errors"
+	"fmt"
+	"os"
 
 	"github.com/xanecs/lighthouse/config"
 	"github.com/xanecs/lighthouse/driver"
@@ -57,8 +59,20 @@ func (h *Hardware) Stop() {
 	}
 }
 
+// Restore all Hardware from database
+func (h *Hardware) Restore(b *broker) {
+	for name, dev := range h.devices {
+		status, err := b.fetchStatus(name)
+		if err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			continue
+		}
+		dev.Restore(status)
+	}
+}
+
 // Listen processes Messages from a channel
-func (h *Hardware) Listen(in chan Message, chanErr chan error) {
+func (h *Hardware) Listen(in chan Message, out chan Status, chanErr chan error) {
 	for {
 		msg := <-in
 		dev := h.devices[msg.Device]
@@ -68,6 +82,8 @@ func (h *Hardware) Listen(in chan Message, chanErr chan error) {
 		}
 		if err := dev.HandleMessage(msg.Action, msg.Params); err != nil {
 			chanErr <- err
+			continue
 		}
+		out <- Status{msg.Device, dev.Status()}
 	}
 }

main.go

@@ -29,13 +29,17 @@ func main() {
 	}
 	defer broker.Close()
 
+	hardware.Restore(broker)
+
 	chanErr := make(chan error)
 	brokerMsg := make(chan string)
 	parsedMsg := make(chan Message)
+	status := make(chan Status)
 
 	go broker.listen(brokerMsg, chanErr)
 	go Parser(brokerMsg, parsedMsg, chanErr)
-	go hardware.Listen(parsedMsg, chanErr)
+	go hardware.Listen(parsedMsg, status, chanErr)
+	go broker.updateStatus(status, chanErr)
 
 	for {
 		err := <-chanErr

message.go

@@ -22,3 +22,9 @@ func Parser(in chan string, out chan Message, errOut chan error) {
 		out <- msg
 	}
 }
+
+// Status represents the state of a device
+type Status struct {
+	Device string
+	Params map[string]string
+}