Merge branch 'shelly'

This adds basic support for Shelly 2PM Gen3 devices.

13 files changed, 456 insertions, 30 deletions

diff --git a/README.md b/README.md
index c9f153a..2b40691 100644
--- a/README.md
+++ b/README.md
@@ -6,14 +6,20 @@ implemented by apps connecting to this central Sia server.
 
 ## Supported vendors and devices
 
-Currently only [Homematic IP][1] as a vendor and the [OpenCCU][2] as interface
-is supported.
+### Homematic IP
 
-The currently only supported device is:
+[Homematic IP][1] with the [OpenCCU][2] as central device is supported with
+these IoT products:
 
 - [HmIP-SWDO-2][3] - optical window or door contact
 
-Further device support is planned.
+### Shelly
+
+[Shelly][8] is supported with the following product:
+
+- [Shelly 2PM Gen3][7] - twin relay e.g. for roller shutter control
+
+The implemented API is documented [here][6].
 
 ## Build instructions
 
@@ -65,8 +71,67 @@ Only aspects explicitly stated here are part of the public API:
 - configuration file format
 - MQTT interface
 
+## MQTT interface
+
+The Sia server connects to a MQTT broker and exposes its client interface
+there.
+
+The MQTT broker host, port and the Sia server's client ID is configured via the
+Sia configuration file.
+
+Furthermore a topic prefix is selected. **All topics documented below are
+implicitly prefixed with this topic prefix.** This allows using multiple Sia
+server instances on one MQTT broker.
+
+All message payloads are UTF-8 encoded strings.
+
+For all terms not explained here see the [MQTT version 3.1.1 documentation][9].
+
+### `/server/health`
+
+- description: Indicates if Sia server is connected to the broker
+- direction: Sia server to client
+- Quality of Service: QoS 1 (at least once)
+- retained: yes
+- receives will message: yes, indicating sudden disconnect of Sia server
+- topic parameters: none
+- payloads:
+    - `good`: Sia server is connected to MQTT broker
+    - `bad`: Sia server is disconnected from MQTT broker
+
+### `/contact/<id>/state`
+
+- description: Indicates state of Homematic IP SWDO-2 contacts
+- direction: Sia server to client
+- Quality of Service: QoS 1 (at least once)
+- retained: yes
+- receives will message: no
+- topic parameters:
+    - `id`: ID of the Homematic IP SWDO-2 contact
+- payloads:
+    - `open`: contact is open
+    - `closed`: contact is closed
+
+### `/cover/<id>/movement`
+
+- description: Allows control of Shelly 2PM Gen3 covers
+- direction: client to Sia server
+- Quality of Service: QoS 2 (exactly once)
+- retained: no
+- receives will message: no
+- topic parameters:
+    - `id`: ID of the Shelly 2PM Gen3 cover
+- payloads:
+    - `extend`: cover increases the covering surface
+    - `retract`: cover decreases the covering surface
+    - `stop`: cover stops current motion if given
+
 [1]: https://homematic-ip.com/
 [2]: https://openccu.de/
 [3]: https://homematic-ip.com/en/product/window-and-door-contact-optical
 [4]: https://systemd.io/
 [5]: https://semver.org/
+[6]: https://shelly-api-docs.shelly.cloud/gen2/
+[7]: https://www.shelly.com/de/products/shelly-2pm-gen3-1/
+[8]: https://shelly.com/
+[9]: https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html
diff --git a/config.go b/config.go
index d39efca..a294b6c 100644
--- a/config.go
+++ b/config.go
@@ -31,7 +31,7 @@ func init() {
 	mqttTopicPrefixRegexp = regexp.MustCompile(MQTT_TOPIC_PREFIX_REGEX)
 }
 
-//go:embed configs/default.json
+//go:embed configs/valid/default.json
 var defaultConfig []byte
 
 type MQTTConfig struct {
@@ -45,9 +45,17 @@ type HomematicConfig struct {
 	PollingPeriod string `json:"polling-period"`
 }
 
+type ShellyConfig struct {
+	ID string `json:"id"`
+	IP string `json:"ip"`
+}
+
+type ShellyConfigs []ShellyConfig
+
 type StartupConfig struct {
 	MQTT      MQTTConfig      `json:"mqtt"`
 	Homematic HomematicConfig `json:"homematic"`
+	Shelly    ShellyConfigs   `json:"shelly"`
 }
 
 func (sc StartupConfig) String() string {
@@ -117,6 +125,13 @@ func (sc StartupConfig) Validate() error {
 		return fmt.Errorf("homematic/polling-period configuration '%s' could not be parsed to duration: %v", sc.Homematic.PollingPeriod, err)
 	}
 
+	for _, shelly := range sc.Shelly {
+		ip := net.ParseIP(shelly.IP)
+		if ip == nil {
+			return fmt.Errorf("Failed to parse IP address '%s'.", shelly.IP)
+		}
+	}
+
 	return nil
 }
 
diff --git a/config_test.go b/config_test.go
index 0972d5d..78d07fb 100644
--- a/config_test.go
+++ b/config_test.go
@@ -1,19 +1,35 @@
 package main
 
 import (
+	"embed"
+	"io/fs"
 	"testing"
 )
 
-func TestDefaultConfig(t *testing.T) {
-	config := StartupConfig{}
+//go:embed configs/valid/*.json
+var valid embed.FS
 
-	err := config.FromJSON(defaultConfig)
-	if err != nil {
-		t.Fatalf("Failed parsing default config from JSON: %v", err)
-	}
+func TestValidConfigs(t *testing.T) {
+	fs.WalkDir(valid, ".", func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			t.Fatalf("Failed to walk valid config files: %v", err)
+		}
 
-	err = config.Validate()
-	if err != nil {
-		t.Fatalf("Failed to validate default config: %v", err)
-	}
+		if d.IsDir() {
+			return nil
+		}
+
+		data, err := valid.ReadFile(path)
+		if err != nil {
+			t.Fatalf("Failed to read config from path %s: %v", path, err)
+		}
+
+		config := StartupConfig{}
+		err = config.FromJSON(data)
+		if err != nil {
+			t.Fatalf("Failed parsing config %s from JSON: %v", path, err)
+		}
+
+		return nil
+	})
 }
diff --git a/configs/meson.build b/configs/meson.build
index 98110da..aa16a47 100644
--- a/configs/meson.build
+++ b/configs/meson.build
@@ -1,3 +1,3 @@
 fs = import('fs')
 
-default_config = fs.copyfile(meson.current_source_dir() / 'default.json')
+default_config = fs.copyfile(meson.current_source_dir() / 'valid' / 'default.json')
diff --git a/configs/default.json b/configs/valid/default.json
index b291185..b291185 100644
--- a/configs/default.json
+++ b/configs/valid/default.json
diff --git a/configs/valid/mqtt-topic-prefix-max-characters.json b/configs/valid/mqtt-topic-prefix-max-characters.json
new file mode 100644
index 0000000..99f3cf1
--- /dev/null
+++ b/configs/valid/mqtt-topic-prefix-max-characters.json
@@ -0,0 +1,11 @@
+{
+	"mqtt": {
+		"broker": "tcp://127.0.0.1:1883",
+		"client-id": "siaserver",
+		"topic-prefix": "aaaaaaaaaaaaaaaaaaaa"
+	},
+	"homematic": {
+		"ccu": "http://127.0.0.1:8080",
+		"polling-period": "50ms"
+	}
+}
diff --git a/configs/valid/shelly.json b/configs/valid/shelly.json
new file mode 100644
index 0000000..578f6dc
--- /dev/null
+++ b/configs/valid/shelly.json
@@ -0,0 +1,21 @@
+{
+	"mqtt": {
+		"broker": "tcp://127.0.0.1:1883",
+		"client-id": "siaserver",
+		"topic-prefix": "sia"
+	},
+	"homematic": {
+		"ccu": "http://127.0.0.1:8080",
+		"polling-period": "50ms"
+	},
+	"shelly": [
+		{
+			"id": "shelly1",
+			"ip": "192.168.1.20"
+		},
+		{
+			"id": "shelly2",
+			"ip": "2001:db8::68"
+		}
+	]
+}
diff --git a/main.go b/main.go
index 6a59d64..1678eac 100644
--- a/main.go
+++ b/main.go
@@ -19,9 +19,11 @@ func main() {
 	config := GetStartupConfig(flags.ConfigPath)
 
 	tx := make(chan MQTTMessage)
+	coverMovement := NewRoute("cover/+/movement", QoS2)
 
-	go MQTTRun(config.MQTT, tx)
+	go MQTTRun(config.MQTT, tx, coverMovement)
 	go HomematicRun(config.Homematic, tx)
+	go ShellyRun(config.Shelly, coverMovement)
 
 	Await(syscall.SIGTERM, syscall.SIGINT)
 }
diff --git a/meson.build b/meson.build
index fb1ec5d..e8878ce 100644
--- a/meson.build
+++ b/meson.build
@@ -3,12 +3,13 @@ project(
   version : '0.1.0-dev',
 )
 
-subdir('configs')
-subdir('systemd')
-
 go = find_program('go', required : true)
 tar = find_program('tar', required : true)
 
+subdir('configs')
+subdir('systemd')
+subdir('tools')
+
 fs = import('fs')
 
 readme = fs.copyfile(meson.current_source_dir() / 'README.md')
@@ -22,6 +23,7 @@ sia_server_linux_amd64 = custom_target(
     meson.current_source_dir() / 'homematic.go',
     meson.current_source_dir() / 'config.go',
     meson.current_source_dir() / 'flags.go',
+    meson.current_source_dir() / 'shelly.go',
   ],
   output : 'sia-server-linux-amd64',
   env : {'GOOS': 'linux', 'GOARCH': 'amd64'},
diff --git a/mqtt.go b/mqtt.go
index a7b374d..fd1cba1 100644
--- a/mqtt.go
+++ b/mqtt.go
@@ -4,14 +4,18 @@ import (
 	"fmt"
 	"log"
 	"time"
+	"strings"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 )
 
 const (
-	QOS                        = byte(1)
+	QoS0                       = byte(0)
+	QoS1                       = byte(1)
+	QoS2                       = byte(2)
 	RETAINED                   = true
 	MQTT_CONNECT_TIMEOUT       = 1 * time.Second
+	MQTT_SUBSCRIBE_TIMEOUT     = 1 * time.Second
 	MQTT_DISCONNECT_TIMEOUT_US = 500
 	MQTT_KEEPALIVE_PERIOD      = 2 * time.Second
 )
@@ -25,20 +29,52 @@ type MQTTMessage struct {
 	Payload []byte
 }
 
-func MQTTRun(config MQTTConfig, tx chan MQTTMessage) {
+type Route struct {
+	Topic       string
+	QoS         byte
+	Destination chan MQTTMessage
+}
+
+func NewRoute(topic string, qos byte) Route {
+	return Route{topic, qos, make(chan MQTTMessage)}
+}
+
+func (m MQTTMessage) String() string {
+	return fmt.Sprintf("topic='%s' message='%s'", m.Topic, string(m.Payload))
+}
+
+func MQTTRun(config MQTTConfig, tx chan MQTTMessage, routes ...Route) {
 	mqttServerHealthTopic = fmt.Sprintf("%s/server/health", config.TopicPrefix)
 
 	opts := mqtt.NewClientOptions()
 	opts.AddBroker(config.Broker)
 	opts.SetClientID(config.ClientID)
 	opts.SetCleanSession(true)
-	opts.SetOnConnectHandler(MQTTOnConnectHandler)
+	opts.SetOnConnectHandler(func(c mqtt.Client) {
+		log.Printf("Connected to MQTT broker.")
+		c.Publish(mqttServerHealthTopic, QoS1, true, []byte(`good`))
+
+		for _, route := range routes {
+			topic := config.TopicPrefix + "/" + route.Topic
+			token := c.Subscribe(topic, route.QoS, func(c mqtt.Client, msg mqtt.Message) {
+				message := MQTTMessage{
+					Topic: strings.TrimPrefix(msg.Topic(), config.TopicPrefix + "/"),
+					Payload: msg.Payload(),
+				}
+				route.Destination <- message
+			})
+			success := token.WaitTimeout(MQTT_SUBSCRIBE_TIMEOUT)
+			if !success {
+				log.Fatalf("Topic subscription failed for topic '%s'", topic)
+			}
+		}
+	})
 	opts.SetConnectionLostHandler(MQTTConnectionLostHandler)
 	opts.SetAutoReconnect(true)
 	opts.SetConnectRetry(true)
 	opts.SetConnectTimeout(MQTT_CONNECT_TIMEOUT)
 	opts.SetKeepAlive(MQTT_KEEPALIVE_PERIOD)
-	opts.SetWill(mqttServerHealthTopic, `bad`, QOS, true)
+	opts.SetWill(mqttServerHealthTopic, `bad`, QoS1, true)
 
 	client := mqtt.NewClient(opts)
 
@@ -52,15 +88,10 @@ func MQTTRun(config MQTTConfig, tx chan MQTTMessage) {
 
 	for message := range tx {
 		topic := fmt.Sprintf("%s/%s", config.TopicPrefix, message.Topic)
-		client.Publish(topic, QOS, RETAINED, message.Payload)
+		client.Publish(topic, QoS1, RETAINED, message.Payload)
 	}
 }
 
-func MQTTOnConnectHandler(c mqtt.Client) {
-	log.Printf("Connected to MQTT broker.")
-	c.Publish(mqttServerHealthTopic, QOS, true, []byte(`good`))
-}
-
 func MQTTConnectionLostHandler(c mqtt.Client, err error) {
 	log.Printf("Connection to MQTT broker lost: %v", err)
 }
diff --git a/shelly.go b/shelly.go
new file mode 100644
index 0000000..508b393
--- /dev/null
+++ b/shelly.go
@@ -0,0 +1,90 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"net"
+	"strings"
+
+	"github.com/gorilla/websocket"
+)
+
+func ShellyRun(config ShellyConfigs, route Route) {
+	for message := range route.Destination {
+		ip, command, err := parseMessage(config, message)
+		if err != nil {
+			log.Println(err)
+			continue
+		}
+
+		err = shellySendCommand(ip, command)
+		if err != nil {
+			log.Printf("Could not send command '%s' to %v: %v", command, ip, err)
+		}
+	}
+}
+
+func parseMessage(config ShellyConfigs, m MQTTMessage) (ip *net.IP, command string, err error) {
+	elements := strings.Split(m.Topic, "/")
+
+	if len(elements) != 3 {
+		return nil, "", fmt.Errorf(
+			"Expected three topic levels but got %d in '%s'.",
+			len(elements), m.Topic,
+		)
+	}
+
+	if elements[0] != "cover" || elements[2] != "movement" {
+		return nil, "", fmt.Errorf("Expected cover/<id>/movement but got: %s", m.Topic)
+	}
+
+	switch string(m.Payload) {
+	case "extend":
+		command = "Cover.Close"
+	case "retract":
+		command = "Cover.Open"
+	case "stop":
+		command = "Cover.Stop"
+	default:
+		return nil, "", fmt.Errorf("Invalid payload '%s'.", m.Payload)
+	}
+
+	id := elements[1]
+
+	for _, c := range config {
+		if c.ID == id {
+			ip := net.ParseIP(c.IP)
+			return &ip, command, nil
+		}
+	}
+
+	return nil, "", fmt.Errorf("Got message for unknown cover '%s'", id)
+}
+
+func shellySendCommand(ip *net.IP, command string) error {
+	template := `
+{
+	"jsonrpc":"2.0",
+	"id": 1,
+	"src":"user_1",
+	"method":"%s",
+	"params": {
+		"id":0
+	}
+}
+`
+	message := fmt.Appendf([]byte{}, template, command)
+
+	c, _, err := websocket.DefaultDialer.Dial("ws://" + ip.String() + "/rpc", nil)
+	if err != nil {
+		return fmt.Errorf("Could not connect to Shelly: %w", err)
+	}
+	defer c.Close()
+
+	err = c.WriteMessage(websocket.TextMessage, message)
+	if err != nil {
+		return fmt.Errorf("Failed writing websocket message to Shelly: %w", err)
+	}
+
+	return nil
+}
diff --git a/tools/meson.build b/tools/meson.build
new file mode 100644
index 0000000..1322ce0
--- /dev/null
+++ b/tools/meson.build
@@ -0,0 +1,15 @@
+websocket_linux_amd64 = custom_target(
+  input : [
+    meson.current_source_dir() / 'websocket.go',
+  ],
+  output : 'websocket-linux-amd64',
+  env : {'GOOS': 'linux', 'GOARCH': 'amd64'},
+  command : [
+    go,
+    'build',
+    '-o',
+    '@OUTPUT@',
+    '@INPUT@',
+  ],
+  build_by_default : true,
+)
diff --git a/tools/websocket.go b/tools/websocket.go
new file mode 100644
index 0000000..af23959
--- /dev/null
+++ b/tools/websocket.go
@@ -0,0 +1,158 @@
+// Websocket debug tool
+//
+// Usage: ./websocket-linux-amd64 ws://<shelly-ip>/rpc
+//
+// This tools is intended to support development of the Websocket-based
+// application programming interface (API) of the Shelly Internet of Things
+// (IoT) devices.
+
+package main
+
+import (
+	"encoding/json"
+	"log"
+	"net/url"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+func main() {
+	log.SetFlags(0)
+
+	interrupt := make(chan os.Signal, 1)
+	signal.Notify(interrupt, os.Interrupt)
+
+	var u url.URL = getURL()
+	log.Printf("connecting to %s", u.String())
+
+	c, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer c.Close()
+
+	go rx(c)
+
+	getConfig(c)
+	coverClose(c)
+	time.Sleep(1 * time.Second)
+	coverOpen(c)
+
+	Await(syscall.SIGTERM, syscall.SIGINT)
+}
+
+func getURL() url.URL {
+	if len(os.Args) != 2 {
+		log.Fatalf("Exactly one argument expected but got %d.", len(os.Args) - 1)
+	}
+
+	maybeURL, err := url.Parse(os.Args[1])
+	if err != nil {
+		log.Fatalf("Cannot parse given URL: %s", os.Args[1])
+	}
+
+	return *maybeURL
+}
+
+func Await(signals ...os.Signal) {
+	listener := make(chan os.Signal, 1)
+	signal.Notify(listener, signals...)
+	defer signal.Stop(listener)
+
+	sig := <-listener
+	log.Printf("Received OS signal '%v'\n", sig)
+}
+
+func getConfig(c *websocket.Conn) {
+	tx(c, `
+{
+	"jsonrpc":"2.0",
+	"id": 1,
+	"src":"user_1",
+	"method":"Sys.GetConfig",
+	"params": {
+		"id":2
+	}
+}
+`)
+}
+
+func coverClose(c *websocket.Conn) {
+	tx(c, `
+{
+	"jsonrpc":"2.0",
+	"id": 1,
+	"src":"user_1",
+	"method":"Cover.Close",
+	"params": {
+		"id":0
+	}
+}
+`)
+}
+
+func coverOpen(c *websocket.Conn) {
+	tx(c, `
+{
+	"jsonrpc":"2.0",
+	"id": 1,
+	"src":"user_1",
+	"method":"Cover.Open",
+	"params": {
+		"id":0
+	}
+}
+`)
+}
+
+func rx(c *websocket.Conn) {
+	for {
+		_, message, err := c.ReadMessage()
+		if err != nil {
+			log.Println("read:", err)
+			return
+		}
+		log.Println("")
+		log.Println(quote(prettify(string(message)), "< "))
+	}
+}
+
+func tx(c *websocket.Conn, d string) {
+	log.Println(quote(prettify(d), "> "))
+
+	err := c.WriteMessage(websocket.TextMessage, []byte(d))
+	if err != nil {
+		log.Fatal(err)
+	}
+}
+
+func prettify(input string) string {
+	var parsed any
+
+	err := json.Unmarshal([]byte(input), &parsed)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	pretty, err := json.MarshalIndent(parsed, "", "    ")
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return string(pretty)
+}
+
+func quote(input string, quotation string) string {
+	lines := strings.Split(input, "\n")
+
+	for i, line := range lines {
+		lines[i] = quotation + line
+	}
+
+	return strings.Join(lines, "\n")
+}