diff options
| -rw-r--r-- | README.md | 73 | ||||
| -rw-r--r-- | config.go | 17 | ||||
| -rw-r--r-- | config_test.go | 36 | ||||
| -rw-r--r-- | configs/meson.build | 2 | ||||
| -rw-r--r-- | configs/valid/default.json (renamed from configs/default.json) | 0 | ||||
| -rw-r--r-- | configs/valid/mqtt-topic-prefix-max-characters.json | 11 | ||||
| -rw-r--r-- | configs/valid/shelly.json | 21 | ||||
| -rw-r--r-- | main.go | 4 | ||||
| -rw-r--r-- | meson.build | 8 | ||||
| -rw-r--r-- | mqtt.go | 51 | ||||
| -rw-r--r-- | shelly.go | 90 | ||||
| -rw-r--r-- | tools/meson.build | 15 | ||||
| -rw-r--r-- | tools/websocket.go | 158 |
13 files changed, 456 insertions, 30 deletions
@@ -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 @@ -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" + } + ] +} @@ -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'}, @@ -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") +} |