More refactoring, made zigbee2mqtt prefix a config option

automation/automation.go

@@ -4,12 +4,39 @@ import (
 	"automation/home"
 	"automation/integration/hue"
 	"automation/integration/ntfy"
+	"encoding/json"
+	"log"
 
 	paho "github.com/eclipse/paho.mqtt.golang"
 )
 
-func RegisterAutomations(client paho.Client, hue *hue.Hue, notify *ntfy.Notify, home *home.Home) {
+func on[M any](client paho.Client, topic string, onMessage func(message M)) {
+	var handler paho.MessageHandler = func(c paho.Client, m paho.Message) {
+		if len(m.Payload()) == 0 {
+			// In this case we clear the persistent message
+			// @TODO Maybe implement onClear as a callback? (Currently not needed)
+			return
+		}
+
+		var message M;
+		err := json.Unmarshal(m.Payload(), &message)
+		if err != nil {
+			log.Println(err)
+			return
+		}
+
+		if onMessage != nil {
+			onMessage(message)
+		}
+	}
+
+	if token := client.Subscribe(topic, 1, handler); token.Wait() && token.Error() != nil {
+		log.Println(token.Error())
+	}
+}
+
+func RegisterAutomations(client paho.Client, prefix string, hue *hue.Hue, notify *ntfy.Notify, home *home.Home) {
 	presenceAutomation(client, hue, notify, home)
 	mixerAutomation(client, home)
-	kettleAutomation(client, home)
+	kettleAutomation(client, prefix, home)
 }

automation/kettle.go

@@ -3,6 +3,7 @@ package automation
 import (
 	"automation/device"
 	"automation/home"
+	"automation/integration/zigbee"
 	"fmt"
 	"log"
 	"time"
@@ -10,29 +11,19 @@ import (
 	paho "github.com/eclipse/paho.mqtt.golang"
 )
 
-func kettleAutomation(client paho.Client, home *home.Home) {
+func kettleAutomation(client paho.Client, prefix string, home *home.Home) {
 	const name = "kitchen/kettle"
 	const length = 5 * time.Minute
 
 	timer := time.NewTimer(length)
 
-	var handler paho.MessageHandler = func(c paho.Client, m paho.Message) {
+	on(client, fmt.Sprintf("%s/%s", prefix, name), func(message zigbee.OnOffState) {
-		kettle, err := device.GetDevice[device.OnOff](&home.Devices, name)
+		if message.State {
-		if err != nil {
-			log.Println(err)
-			return
-		}
-
-		if kettle.GetOnOff() {
 			timer.Reset(length)
 		} else {
 			timer.Stop()
 		}
-	}
+	})
-
-	if token := client.Subscribe(fmt.Sprintf("zigbee2mqtt/%s", name), 1, handler); token.Wait() && token.Error() != nil {
-		log.Println(token.Error())
-	}
 
 	go func() {
 		for {

automation/mixer.go

@@ -3,14 +3,14 @@ package automation
 import (
 	"automation/device"
 	"automation/home"
-	"encoding/json"
+	"automation/integration/zigbee"
 	"log"
 
 	paho "github.com/eclipse/paho.mqtt.golang"
 )
 
 func mixerAutomation(client paho.Client, home *home.Home) {
-	var handler paho.MessageHandler = func(client paho.Client, msg paho.Message) {
+	on(client, "test/remote", func(message zigbee.RemoteState) {
 		mixer, err := device.GetDevice[device.OnOff](&home.Devices, "living_room/mixer")
 		if err != nil {
 			log.Println(err)
@@ -22,23 +22,14 @@ func mixerAutomation(client paho.Client, home *home.Home) {
 			return
 		}
 
-		var message struct {
+		if message.Action == zigbee.ACTION_ON {
-			Action string `json:"action"`
-		}
-		err = json.Unmarshal(msg.Payload(), &message)
-		if err != nil {
-			log.Println(err)
-			return
-		}
-
-		if message.Action == "on" {
 			if mixer.GetOnOff() {
 				mixer.SetOnOff(false)
 				speakers.SetOnOff(false)
 			} else {
 				mixer.SetOnOff(true)
 			}
-		} else if message.Action == "brightness_move_up" {
+		} else if message.Action == zigbee.ACTION_BRIGHTNESS_UP {
 			if speakers.GetOnOff() {
 				speakers.SetOnOff(false)
 			} else {
@@ -46,10 +37,6 @@ func mixerAutomation(client paho.Client, home *home.Home) {
 				mixer.SetOnOff(true)
 			}
 		}
-	}
+	})
-
-	if token := client.Subscribe("test/remote", 1, handler); token.Wait() && token.Error() != nil {
-		log.Println(token.Error())
-	}
 }
 

automation/presence.go

@@ -6,7 +6,6 @@ import (
 	"automation/integration/hue"
 	"automation/integration/ntfy"
 	"automation/presence"
-	"encoding/json"
 	"fmt"
 	"log"
 
@@ -14,18 +13,7 @@ import (
 )
 
 func presenceAutomation(client paho.Client, hue *hue.Hue, notify *ntfy.Notify, home *home.Home) {
-	var handler paho.MessageHandler = func(client paho.Client, msg paho.Message) {
+	on(client, "automation/presence", func(message presence.Message) {
-		if len(msg.Payload()) == 0 {
-			// In this case we clear the persistent message
-			return
-		}
-		var message presence.Message
-		err := json.Unmarshal(msg.Payload(), &message)
-		if err != nil {
-			log.Println(err)
-			return
-		}
-
 		fmt.Printf("Presence: %t\n", message.State)
 
 		// Set presence on the hue bridge
@@ -51,9 +39,5 @@ func presenceAutomation(client paho.Client, hue *hue.Hue, notify *ntfy.Notify, h
 
 		// Notify users of presence update
 		notify.Presence(message.State)
-	}
+	})
-
-	if token := client.Subscribe("automation/presence", 1, handler); token.Wait() && token.Error() != nil {
-		log.Println(token.Error())
-	}
 }

config.yml

@@ -7,6 +7,9 @@ mqtt:
   username: mqtt
   client_id: automation
 
+zigbee:
+  prefix: zigbee2mqtt
+
 kasa:
   outlets:
     living_room/mixer: 10.0.0.49

config/config.go

@@ -28,6 +28,10 @@ type config struct {
 		ClientID string `yaml:"client_id" envconfig:"MQTT_CLIENT_ID"`
 	} `yaml:"mqtt"`
 
+	Zigbee struct {
+		MQTTPrefix string `yaml:"prefix" envconfig:"ZIGBEE2MQTT_PREFIX"`
+	}
+
 	Kasa struct {
 		Outlets map[device.InternalName]string `yaml:"outlets"`
 	} `yaml:"kasa"`

integration/zigbee/devices.go

@@ -5,12 +5,13 @@ import (
 	"automation/home"
 	"context"
 	"encoding/json"
+	"fmt"
 	"log"
 
 	paho "github.com/eclipse/paho.mqtt.golang"
 )
 
-func DevicesHandler(client paho.Client, home *home.Home) {
+func DevicesHandler(client paho.Client, prefix string, home *home.Home) {
 	var handler paho.MessageHandler = func(client paho.Client, msg paho.Message) {
 		var devices []Info
 		json.Unmarshal(msg.Payload(), &devices)
@@ -24,6 +25,7 @@ func DevicesHandler(client paho.Client, home *home.Home) {
 		for _, d := range devices {
 			switch d.Description {
 			case "Kettle":
+				d.MQTTAddress = fmt.Sprintf("%s/%s", prefix, d.FriendlyName.String())
 				kettle := NewKettle(d, client, home.Service)
 				home.AddDevice(kettle)
 			}
@@ -35,7 +37,7 @@ func DevicesHandler(client paho.Client, home *home.Home) {
 		home.Service.RequestSync(context.Background(), home.Username)
 	}
 
-	if token := client.Subscribe("zigbee2mqtt/bridge/devices", 1, handler); token.Wait() && token.Error() != nil {
+	if token := client.Subscribe(fmt.Sprintf("%s/bridge/devices", prefix), 1, handler); token.Wait() && token.Error() != nil {
 		log.Println(token.Error())
 	}
 }

integration/zigbee/kettle.go

@@ -27,7 +27,7 @@ type kettle struct {
 func NewKettle(info Info, client paho.Client, service *google.Service) *kettle {
 	k := &kettle{info: info, client: client, service: service, updated: make(chan bool, 1)}
 
-	if token := k.client.Subscribe(fmt.Sprintf("zigbee2mqtt/%s", k.info.FriendlyName), 1, k.stateHandler); token.Wait() && token.Error() != nil {
+	if token := k.client.Subscribe(k.info.MQTTAddress, 1, k.stateHandler); token.Wait() && token.Error() != nil {
 		log.Println(token.Error())
 	}
 
@@ -35,13 +35,11 @@ func NewKettle(info Info, client paho.Client, service *google.Service) *kettle {
 }
 
 func (k *kettle) stateHandler(client paho.Client, msg paho.Message) {
-	var payload struct {
+	var payload OnOffState
-		State string `json:"state"`
-	}
 	json.Unmarshal(msg.Payload(), &payload)
 
 	// Update the internal state
-	k.isOn = payload.State == "ON"
+	k.isOn = payload.State
 	k.online = true
 
 	// Notify that the state has updated
@@ -68,7 +66,7 @@ func (k *kettle) IsZigbeeDevice() {}
 
 // zigbee.Device
 func (k *kettle) Delete() {
-	if token := k.client.Unsubscribe(fmt.Sprintf("zigbee2mqtt/%s", k.info.FriendlyName)); token.Wait() && token.Error() != nil {
+	if token := k.client.Unsubscribe(k.info.MQTTAddress); token.Wait() && token.Error() != nil {
 		log.Println(token.Error())
 	}
 }
@@ -167,7 +165,7 @@ func (k *kettle) SetOnOff(state bool) {
 		msg = "ON"
 	}
 
-	if token := k.client.Publish(fmt.Sprintf("zigbee2mqtt/%s/set", k.info.FriendlyName), 1, false, fmt.Sprintf(`{ "state": "%s" }`, msg)); token.Wait() && token.Error() != nil {
+	if token := k.client.Publish(fmt.Sprintf("%s/set", k.info.MQTTAddress), 1, false, fmt.Sprintf(`{ "state": "%s" }`, msg)); token.Wait() && token.Error() != nil {
 		log.Println(token.Error())
 	}
 }

integration/zigbee/payload.go

@@ -0,0 +1,34 @@
+package zigbee
+
+import "encoding/json"
+
+type OnOffState struct {
+	State bool
+}
+
+func (k *OnOffState) UnmarshalJSON(data []byte) error {
+	var payload struct {
+		State string `json:"state"`
+	}
+	if err := json.Unmarshal(data, &payload); err != nil {
+		return err
+	}
+
+	k.State = payload.State == "ON"
+
+	return nil
+}
+
+type RemoteAction string
+
+const (
+	ACTION_ON              RemoteAction = "on"
+	ACTION_OFF                          = "off"
+	ACTION_BRIGHTNESS_UP                = "brightness_move_up"
+	ACTION_BRIGHTNESS_DOWN              = "brightness_move_down"
+	ACTION_BRIGHTNESS_STOP              = "brightness_move_down"
+)
+
+type RemoteState struct {
+	Action RemoteAction `json:"action"`
+}

integration/zigbee/zigbee.go

@@ -9,6 +9,8 @@ type Info struct {
 	Manufacturer    string       `json:"manufacturer"`
 	ModelID         string       `json:"model_id"`
 	SoftwareBuildID string       `json:"software_build_id"`
+
+	MQTTAddress string `json:"-"`
 }
 
 type Device interface {

main.go

@@ -53,7 +53,7 @@ func main() {
 	}
 	defer client.Disconnect(250)
 
-	zigbee.DevicesHandler(client, home)
+	zigbee.DevicesHandler(client, cfg.Zigbee.MQTTPrefix, home)
 
 	p := presence.New(client, hue, notify, home)
 	defer p.Delete(client)
@@ -65,7 +65,7 @@ func main() {
 	}
 	defer automationClient.Disconnect(250)
 
-	automation.RegisterAutomations(automationClient, hue, notify, home)
+	automation.RegisterAutomations(automationClient, cfg.Zigbee.MQTTPrefix, hue, notify, home)
 
 	addr := ":8090"
 	srv := http.Server{