From 42ec94d9c71225c395093b6ef418120f93b4f4ae Mon Sep 17 00:00:00 2001 From: Eddine97 Date: Sat, 17 Oct 2026 01:29:32 +0200 Subject: [PATCH 1/4] feat(api,mysensors): add action payload and detect bootloader --- cmd/client/api/action.go | 16 +- pkg/api/action/api.go | 4 +- pkg/http_router/routes/action.go | 8 +- .../mysensors_v2/bootloader_detect.go | 155 ++++++++++++++++++ .../mysensors_v2/bootloader_detect_test.go | 63 +++++++ 5 files changed, 239 insertions(+), 7 deletions(-) create mode 100644 plugin/gateway/provider/mysensors_v2/bootloader_detect.go create mode 100644 plugin/gateway/provider/mysensors_v2/bootloader_detect_test.go diff --git a/cmd/client/api/action.go b/cmd/client/api/action.go index 76c70e1a..49dbcfdc 100644 --- a/cmd/client/api/action.go +++ b/cmd/client/api/action.go @@ -7,15 +7,20 @@ import ( webHandlerTY "github.com/mycontroller-org/server/v2/pkg/types/web_handler" ) -func (c *Client) ExecuteNodeAction(action string, ids []string) error { - return c.executeIDAction(API_ACTION_NODE, action, ids) +// ExecuteNodeAction sends an action to one or more nodes. +func (c *Client) ExecuteNodeAction(action string, ids []string, payload ...string) error { + var pl string + if len(payload) > 0 { + pl = payload[0] + } + return c.executeIDAction(API_ACTION_NODE, action, pl, ids) } func (c *Client) ExecuteGatewayAction(action string, ids []string) error { - return c.executeIDAction(API_ACTION_GATEWAY, action, ids) + return c.executeIDAction(API_ACTION_GATEWAY, action, "", ids) } -func (c *Client) executeIDAction(api, action string, ids []string) error { +func (c *Client) executeIDAction(api, action, payload string, ids []string) error { if action == "" { return fmt.Errorf("action is required") } @@ -26,6 +31,9 @@ func (c *Client) executeIDAction(api, action string, ids []string) error { "action": action, "id": ids, } + if payload != "" { + query["payload"] = payload + } _, err := c.executeJson(api, http.MethodGet, nil, query, nil, http.StatusOK) return err } diff --git a/pkg/api/action/api.go b/pkg/api/action/api.go index 91538ac3..3e63d5c9 100644 --- a/pkg/api/action/api.go +++ b/pkg/api/action/api.go @@ -57,7 +57,7 @@ func (a *ActionAPI) ExecuteActionOnResourceByQuickID(data *handlerTY.ResourceDat case quickIdUtils.QuickIdNode: gatewayID := kvMap[types.KeyGatewayID] nodeID := kvMap[types.KeyNodeID] - return a.toNode(nil, gatewayID, nodeID, data.Payload) + return a.toNode(nil, gatewayID, nodeID, data.Payload, "") case quickIdUtils.QuickIdSource: // no action needed @@ -120,7 +120,7 @@ func (a *ActionAPI) ExecuteActionOnResourceByLabels(data *handlerTY.ResourceData items := result.Data.(*[]nodeTY.Node) for index := 0; index < len(*items); index++ { item := (*items)[index] - err = a.toNode(&item, item.GatewayID, item.NodeID, data.Payload) + err = a.toNode(&item, item.GatewayID, item.NodeID, data.Payload, "") if err != nil { a.logger.Error("error on sending data", zap.Error(err), zap.String("nodeID", item.ID), zap.String("payload", data.Payload)) } diff --git a/pkg/http_router/routes/action.go b/pkg/http_router/routes/action.go index 5b414852..c1c64cb3 100644 --- a/pkg/http_router/routes/action.go +++ b/pkg/http_router/routes/action.go @@ -38,7 +38,13 @@ func (h *Routes) executeNodeAction(w http.ResponseWriter, r *http.Request) { return } - err = h.action.ExecuteNodeAction(actionArr[0], idsArr) + // Optional payload for actions that require one (e.g. SetSleepDuration) + payload := "" + if payloadArr := query[keyPayload]; len(payloadArr) > 0 { + payload = payloadArr[0] + } + + err = h.action.ExecuteNodeAction(actionArr[0], payload, idsArr) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return diff --git a/plugin/gateway/provider/mysensors_v2/bootloader_detect.go b/plugin/gateway/provider/mysensors_v2/bootloader_detect.go new file mode 100644 index 00000000..fd7d16c1 --- /dev/null +++ b/plugin/gateway/provider/mysensors_v2/bootloader_detect.go @@ -0,0 +1,155 @@ +package mysensors + +import ( + "encoding/binary" + hexENC "encoding/hex" + "fmt" + "strings" + + nodeTY "github.com/mycontroller-org/server/v2/pkg/types/node" + "go.uber.org/zap" +) + +// Labels set on nodes to remember the detected bootloader. +const ( + LabelBootloaderKind = "bootloader_kind" // e.g. "mysbootloader-1.4", "protocol-3.1" + LabelBootloaderVersion = "bootloader_version" // e.g. "1.4" (MYSBootloader only) +) + +// Bootloader kinds recognised by the controller. +const ( + BootloaderKindMYSBootloader10 = "mysbootloader-1.0" + BootloaderKindMYSBootloader13 = "mysbootloader-1.3" + BootloaderKindMYSBootloader14 = "mysbootloader-1.4" + BootloaderKindProtocol31 = "protocol-3.1" + BootloaderKindUnknown = "unknown" +) + +// isMYSBootloaderKind reports whether a kind string designates any MYSBootloader. +func isMYSBootloaderKind(kind string) bool { + return strings.HasPrefix(kind, "mysbootloader") +} + +// detectBootloaderKind classifies the bootloader from a ST_FIRMWARE_CONFIG_REQUEST +// payload. Uses payload length and (for the 10-byte format) the MYSBootloader +// BLVersion field. +// +// Signatures: +// +// 6 bytes : MYSBootloader < 1.3 (type, version, blocks) +// 10 bytes : MYSBootloader >= 1.3 (type, version, blocks, crc, BLVersion) +// 11 bytes : protocol 3.1 (10 bytes + blockSize) +// 12+ bytes: protocol 3.1 extended (custom) +func detectBootloaderKind(requestHex string) string { + b, err := hexENC.DecodeString(sanitizeOtaHex(requestHex)) + if err != nil || len(b) == 0 { + return "" + } + switch len(b) { + case 6: + return BootloaderKindMYSBootloader10 + case 10: + // BLVersion is a little-endian uint16 at offset 8. + // MYSBootloader packs (minor << 8) | major. + blv := binary.LittleEndian.Uint16(b[8:10]) + major := blv & 0xFF + minor := (blv >> 8) & 0xFF + if major == 1 { + switch minor { + case 3: + return BootloaderKindMYSBootloader13 + case 4: + return BootloaderKindMYSBootloader14 + default: + if minor >= 3 && minor <= 9 { + return fmt.Sprintf("mysbootloader-1.%d", minor) + } + return "mysbootloader-compatible" + } + } + return BootloaderKindUnknown + case 11, 12, 18: + // Protocol 3.1: at offset 10, blockSize (must be a positive multiple of 8). + if b[10] >= 8 && b[10]%8 == 0 { + return BootloaderKindProtocol31 + } + return BootloaderKindUnknown + default: + if len(b) >= 10 && b[10] >= 8 && b[10]%8 == 0 { + return BootloaderKindProtocol31 + } + return BootloaderKindUnknown + } +} + +// extractBootloaderVersion returns a human string like "1.4" for MYSBootloader, +// or "" if the version cannot be determined. +func extractBootloaderVersion(requestHex string) string { + b, err := hexENC.DecodeString(sanitizeOtaHex(requestHex)) + if err != nil || len(b) < 10 { + return "" + } + if len(b) != 10 && len(b) != 11 && len(b) != 12 && len(b) != 18 { + return "" + } + blv := binary.LittleEndian.Uint16(b[8:10]) + major := blv & 0xFF + minor := (blv >> 8) & 0xFF + if major == 0 && minor == 0 { + return "" + } + return fmt.Sprintf("%d.%d", major, minor) +} + +// rememberBootloaderKind persists the detected bootloader in the node labels. +// Does nothing if nothing changed, to avoid unnecessary writes. +func (p *Provider) rememberBootloaderKind(node *nodeTY.Node, requestHex string) { + if node == nil || node.Labels == nil || requestHex == "" { + return + } + kind := detectBootloaderKind(requestHex) + if kind == "" { + return + } + version := extractBootloaderVersion(requestHex) + + changed := false + if node.Labels.Get(LabelBootloaderKind) != kind { + node.Labels.Set(LabelBootloaderKind, kind) + changed = true + } + if version != "" && node.Labels.Get(LabelBootloaderVersion) != version { + node.Labels.Set(LabelBootloaderVersion, version) + changed = true + } + if !changed { + return + } + p.setNodeLabels(node) + p.logger.Info("detected bootloader", + zap.String("nodeId", node.NodeID), + zap.String("gatewayId", node.GatewayID), + zap.String("kind", kind), + zap.String("version", version), + zap.Int("payloadBytes", len(requestHex)/2), + ) +} + +// bootloaderKindForNode returns the persisted bootloader kind for a node, or "". +func bootloaderKindForNode(node *nodeTY.Node) string { + if node == nil || node.Labels == nil { + return "" + } + return node.Labels.Get(LabelBootloaderKind) +} + +// IsMYSBootloader reports whether the node's persisted bootloader kind +// designates any MYSBootloader variant. +func (p *Provider) IsMYSBootloader(node *nodeTY.Node) bool { + return isMYSBootloaderKind(bootloaderKindForNode(node)) +} + +// BootloaderKind returns the persisted bootloader kind for a node, or "". +func (p *Provider) BootloaderKind(node *nodeTY.Node) string { + return bootloaderKindForNode(node) +} diff --git a/plugin/gateway/provider/mysensors_v2/bootloader_detect_test.go b/plugin/gateway/provider/mysensors_v2/bootloader_detect_test.go new file mode 100644 index 00000000..2b825df3 --- /dev/null +++ b/plugin/gateway/provider/mysensors_v2/bootloader_detect_test.go @@ -0,0 +1,63 @@ +package mysensors + +import "testing" + +func TestDetectBootloaderKind(t *testing.T) { + tests := []struct { + name string + hex string + want string + }{ + {"MYSBootloader 1.0 (6 bytes)", "E70019004802", BootloaderKindMYSBootloader10}, + {"MYSBootloader 1.3", "E700190048025B0E0103", BootloaderKindMYSBootloader13}, + {"MYSBootloader 1.4", "E700190048025B0E0104", BootloaderKindMYSBootloader14}, + {"Protocol 3.1 (11 bytes)", "E700190048025B0E010310", BootloaderKindProtocol31}, + {"Protocol 3.1 extended (18 bytes)", "E700190048025B0E010310A10000000000AB000000", BootloaderKindProtocol31}, + {"unknown 10-byte", "E700190048025B0E00FF", BootloaderKindUnknown}, + {"empty", "", ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := detectBootloaderKind(tt.hex) + if got != tt.want { + t.Errorf("detectBootloaderKind(%q) = %q, want %q", tt.hex, got, tt.want) + } + }) + } +} + +func TestExtractBootloaderVersion(t *testing.T) { + tests := []struct { + name string + hex string + want string + }{ + {"1.3", "E700190048025B0E0103", "1.3"}, + {"1.4", "E700190048025B0E0104", "1.4"}, + {"6 bytes no version", "E70019004802", ""}, + {"empty", "", ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := extractBootloaderVersion(tt.hex) + if got != tt.want { + t.Errorf("extractBootloaderVersion(%q) = %q, want %q", tt.hex, got, tt.want) + } + }) + } +} + +func TestIsMYSBootloaderKind(t *testing.T) { + if !isMYSBootloaderKind("mysbootloader-1.4") { + t.Fatal("1.4 must be MYSBootloader") + } + if !isMYSBootloaderKind("mysbootloader-1.0") { + t.Fatal("1.0 must be MYSBootloader") + } + if isMYSBootloaderKind("protocol-3.1") { + t.Fatal("protocol-3.1 must not be MYSBootloader") + } + if isMYSBootloaderKind("unknown") { + t.Fatal("unknown must not be MYSBootloader") + } +} From 457696c2ebdcd592477413532cad0dd97c1d8b2e Mon Sep 17 00:00:00 2001 From: Eddine97 Date: Sat, 17 Oct 2026 01:29:48 +0200 Subject: [PATCH 2/4] feat(mysensors): add set-sleep-duration and pre-OTA reboot --- cmd/client/command/action/cmd.go | 40 +++- cmd/client/command/action/cmd_test.go | 19 +- pkg/api/action/actions_node.go | 17 +- pkg/types/node/types.go | 1 + .../gateway/provider/mysensors_v2/actions.go | 50 ++++- .../gateway/provider/mysensors_v2/constant.go | 43 ++-- .../provider/mysensors_v2/msg_parser.go | 18 +- .../gateway/provider/mysensors_v2/ota_impl.go | 5 + .../provider/mysensors_v2/ota_reboot.go | 166 ++++++++++++++ .../provider/mysensors_v2/ota_reboot_test.go | 85 ++++++++ .../provider/mysensors_v2/ota_store.go | 23 +- .../gateway/provider/mysensors_v2/provider.go | 18 +- .../provider/mysensors_v2/sleep_command.go | 204 ++++++++++++++++++ .../mysensors_v2/sleep_command_test.go | 123 +++++++++++ 14 files changed, 765 insertions(+), 47 deletions(-) create mode 100644 plugin/gateway/provider/mysensors_v2/ota_reboot.go create mode 100644 plugin/gateway/provider/mysensors_v2/ota_reboot_test.go create mode 100644 plugin/gateway/provider/mysensors_v2/sleep_command.go create mode 100644 plugin/gateway/provider/mysensors_v2/sleep_command_test.go diff --git a/cmd/client/command/action/cmd.go b/cmd/client/command/action/cmd.go index 108c224b..fcc100fa 100644 --- a/cmd/client/command/action/cmd.go +++ b/cmd/client/command/action/cmd.go @@ -21,10 +21,12 @@ var actionCmd = &cobra.Command{ Short: "Send an action to a node or gateway", Long: `Send a server action to a node or gateway. -Node actions: reboot, reset, firmware-update, heartbeat, refresh-node-info +Node actions: reboot, reset, firmware-update, heartbeat, refresh-node-info, set-sleep-duration:SENSOR_ID:PERCENT, set-sleep-duration:PERCENT Gateway actions: discover-nodes myc action node reboot mysensor.1 + myc action node set-sleep-duration:206:32 mysensor.1 + myc action node set-sleep-duration:32 mysensor.1 myc action gateway discover-nodes mysensor `, SilenceUsage: true, @@ -42,7 +44,9 @@ var nodeActionCmd = &cobra.Command{ myc action node reset mysensor.1 myc action node firmware-update mysensor.1 myc action node heartbeat mysensor.1 - myc action node refresh-node-info mysensor.1`, + myc action node refresh-node-info mysensor.1 + myc action node set-sleep-duration:206:32 mysensor.1 + myc action node set-sleep-duration:32 mysensor.1`, Args: cobra.MinimumNArgs(3), SilenceUsage: true, SilenceErrors: true, @@ -51,14 +55,36 @@ var nodeActionCmd = &cobra.Command{ }, RunE: func(cmd *cobra.Command, args []string) error { client, rest := rootCmd.TakeAlias(args) - action, err := normalizeNodeAction(rest[0]) + + // Support "action" and "action:payload" + raw := rest[0] + actionStr := raw + payload := "" + if idx := strings.Index(raw, ":"); idx >= 0 { + actionStr = raw[:idx] + payload = raw[idx+1:] + } + + action, err := normalizeNodeAction(actionStr) if err != nil { return err } + + // Payload is required for some actions + if action == nodeTY.ActionSetSleepDuration && payload == "" { + return fmt.Errorf("action %q requires a payload (e.g. set-sleep-duration:32 or set-sleep-duration:206:32)", actionStr) + } + ids, resolveErr := client.ResolveNodeIDs(rest[1:]) if len(ids) > 0 { - if err := client.ExecuteNodeAction(action, ids); err != nil { - return err + if payload != "" { + if err := client.ExecuteNodeAction(action, ids, payload); err != nil { + return err + } + } else { + if err := client.ExecuteNodeAction(action, ids); err != nil { + return err + } } _, _ = fmt.Fprintf(rootCmd.IOStreams.Out, "sent %s to %d node(s)\n", action, len(ids)) } @@ -106,8 +132,10 @@ func normalizeNodeAction(action string) (string, error) { return nodeTY.ActionHeartbeatRequest, nil case "refresh-node-info", "refresh": return nodeTY.ActionRefreshNodeInfo, nil + case "set-sleep-duration": + return nodeTY.ActionSetSleepDuration, nil default: - return "", fmt.Errorf("unsupported node action %q (supported: reboot, reset, firmware-update, heartbeat, refresh-node-info)", action) + return "", fmt.Errorf("unsupported node action %q (supported: reboot, reset, firmware-update, heartbeat, refresh-node-info, set-sleep-duration)", action) } } diff --git a/cmd/client/command/action/cmd_test.go b/cmd/client/command/action/cmd_test.go index 836c6eef..3f7ba857 100644 --- a/cmd/client/command/action/cmd_test.go +++ b/cmd/client/command/action/cmd_test.go @@ -11,15 +11,16 @@ import ( func TestNormalizeNodeAction(t *testing.T) { cases := map[string]string{ - "reboot": nodeTY.ActionReboot, - "reset": nodeTY.ActionReset, - "firmware-update": nodeTY.ActionFirmwareUpdate, - "firmware_update": nodeTY.ActionFirmwareUpdate, - "heartbeat": nodeTY.ActionHeartbeatRequest, - "heartbeat-request": nodeTY.ActionHeartbeatRequest, - "refresh": nodeTY.ActionRefreshNodeInfo, - "refresh-node-info": nodeTY.ActionRefreshNodeInfo, - "refresh_node_info": nodeTY.ActionRefreshNodeInfo, + "reboot": nodeTY.ActionReboot, + "reset": nodeTY.ActionReset, + "firmware-update": nodeTY.ActionFirmwareUpdate, + "firmware_update": nodeTY.ActionFirmwareUpdate, + "heartbeat": nodeTY.ActionHeartbeatRequest, + "heartbeat-request": nodeTY.ActionHeartbeatRequest, + "refresh": nodeTY.ActionRefreshNodeInfo, + "refresh-node-info": nodeTY.ActionRefreshNodeInfo, + "refresh_node_info": nodeTY.ActionRefreshNodeInfo, + "set-sleep-duration": nodeTY.ActionSetSleepDuration, } for in, want := range cases { got, err := normalizeNodeAction(in) diff --git a/pkg/api/action/actions_node.go b/pkg/api/action/actions_node.go index 2a47a230..94b76bd5 100644 --- a/pkg/api/action/actions_node.go +++ b/pkg/api/action/actions_node.go @@ -10,14 +10,15 @@ import ( ) // ExecuteNodeAction for a node -func (a *ActionAPI) ExecuteNodeAction(action string, nodeIDs []string) error { +func (a *ActionAPI) ExecuteNodeAction(action, payload string, nodeIDs []string) error { // verify is a valid action? switch action { case nodeTY.ActionFirmwareUpdate, nodeTY.ActionHeartbeatRequest, nodeTY.ActionReboot, nodeTY.ActionRefreshNodeInfo, - nodeTY.ActionReset: + nodeTY.ActionReset, + nodeTY.ActionSetSleepDuration: // nothing to do, just continue default: return fmt.Errorf("invalid node action:%s", action) @@ -29,15 +30,18 @@ func (a *ActionAPI) ExecuteNodeAction(action string, nodeIDs []string) error { } for index := 0; index < len(nodes); index++ { node := nodes[index] - err = a.toNode(&node, node.GatewayID, node.NodeID, action) + err = a.toNode(&node, node.GatewayID, node.NodeID, action, payload) if err != nil { - a.logger.Error("error on sending an action to a node", zap.Error(err), zap.String("gateway", node.GatewayID), zap.String("node", node.NodeID)) + a.logger.Error("error on sending an action to a node", + zap.Error(err), + zap.String("gateway", node.GatewayID), + zap.String("node", node.NodeID)) } } return nil } -func (a *ActionAPI) toNode(node *nodeTY.Node, gatewayID, nodeID, action string) error { +func (a *ActionAPI) toNode(node *nodeTY.Node, gatewayID, nodeID, action, payload string) error { msg := msgTY.NewMessage(false) msg.GatewayID = gatewayID msg.NodeID = nodeID @@ -58,6 +62,9 @@ func (a *ActionAPI) toNode(node *nodeTY.Node, gatewayID, nodeID, action string) } pl := msgTY.NewPayload() pl.Key = action + if payload != "" { + pl.SetValue(payload) + } msg.Payloads = append(msg.Payloads, pl) msg.Type = msgTY.TypeAction return a.Post(&msg) diff --git a/pkg/types/node/types.go b/pkg/types/node/types.go index 09ccdf0d..b0479bea 100644 --- a/pkg/types/node/types.go +++ b/pkg/types/node/types.go @@ -15,6 +15,7 @@ const ( ActionRefreshNodeInfo = "refresh_node_info" ActionReset = "reset" ActionAwake = "awake" + ActionSetSleepDuration = "SetSleepDuration" ) // Node struct diff --git a/plugin/gateway/provider/mysensors_v2/actions.go b/plugin/gateway/provider/mysensors_v2/actions.go index 92f35601..863b5bac 100644 --- a/plugin/gateway/provider/mysensors_v2/actions.go +++ b/plugin/gateway/provider/mysensors_v2/actions.go @@ -75,7 +75,30 @@ func (p *Provider) handleActions(gwCfg *gwTY.Config, fn string, msg *msgTY.Messa msMsg.Payload = p.getTimestamp(gwCfg) msMsg.Type = actionTime - case nodeTY.ActionFirmwareUpdate, "ST_FIRMWARE_CONFIG_REQUEST": + case nodeTY.ActionFirmwareUpdate: + // User-triggered action: optionally reboot the node before serving + // the OTA config. Required for MYSBootloader-based nodes. + if err := p.maybeSendRebootForOta(msg); err != nil { + p.logger.Error("firmware-update: pre-OTA reboot failed", + zap.String("nodeId", msg.NodeID), + zap.Error(err), + ) + // Continue: the config will be served anyway. + } + pl, err := p.executeFirmwareConfigRequest(msg) + if err != nil { + return err + } + if pl == "" { + return nil + } + msMsg.Command = cmdStream + msMsg.Type = actionFirmwareConfigResponse + msMsg.Payload = strings.ToUpper(pl) + + case "ST_FIRMWARE_CONFIG_REQUEST": + // The node already asked for its config (it is in bootloader mode). + // No reboot needed. pl, err := p.executeFirmwareConfigRequest(msg) if err != nil { return err @@ -99,13 +122,36 @@ func (p *Provider) handleActions(gwCfg *gwTY.Config, fn string, msg *msgTY.Messa msMsg.Type = actionFirmwareResponse msMsg.Payload = strings.ToUpper(pl) + case nodeTY.ActionSetSleepDuration: + raw := strings.TrimSpace(msg.Payloads[0].Value.String()) + sensorID, pct, err := parseSleepDurationPayload(raw) + if err != nil { + return err + } + node, err := p.getNode(msg.GatewayID, msg.NodeID) + if err != nil { + return err + } + if sensorID == "" { + sensorID = node.Labels.Get(LabelSleepCommandSensorID) + } + if sensorID == "" { + return fmt.Errorf("no sleep command sensor configured for node %s; run once: myc action node set-sleep-duration::%d %s", + node.NodeID, pct, msg.GatewayID+"."+msg.NodeID) + } + if sensorID != node.Labels.Get(LabelSleepCommandSensorID) { + node.Labels.Set(LabelSleepCommandSensorID, sensorID) + p.setNodeLabels(node) + } + return p.sendSleepCommand(node, sensorID, pct, "") + default: return fmt.Errorf("this function is not implemented: %s", fn) } return nil } -// geTimestamp returns timestamp in seconds from 1970 +// getTimestamp returns timestamp in seconds from 1970 // adds zone offset to the actual timestamp // user can specify different timezone as a gateway label // if non set, take system timezone diff --git a/plugin/gateway/provider/mysensors_v2/constant.go b/plugin/gateway/provider/mysensors_v2/constant.go index 89097d05..bf58f405 100644 --- a/plugin/gateway/provider/mysensors_v2/constant.go +++ b/plugin/gateway/provider/mysensors_v2/constant.go @@ -5,22 +5,24 @@ import ( "github.com/mycontroller-org/server/v2/pkg/types" msgTY "github.com/mycontroller-org/server/v2/pkg/types/message" + nodeTY "github.com/mycontroller-org/server/v2/pkg/types/node" metricTY "github.com/mycontroller-org/server/v2/plugin/database/metric/types" ) // Labels and fields used in this provider const ( - LabelImperialSystem = "ms_imperial_system" // this is bool label, used to configure about the system metric or imperial - LabelNodeID = "ms_node_id" // MySensors node id - LabelSensorID = "ms_sensor_id" // MySensors sensor id - LabelType = "ms_type" // MySensors type reference, can be used for sensor fields - LabelTypeString = "ms_type_string" // MySensors type in string format - LabelNodeType = "ms_node_type" // MySensors node type - LabelLockedReason = "ms_locked_reason" // If the the node is locked, reason will be in this label - LabelEraseEEPROM = "ms_erase_eeprom" // Supports only for MYSBootloader, on a reboot of node, the node eeprom will be erased - LabelFirmwareTypeID = "ms_type_id" // MySensors firmware type id - LabelFirmwareVersionID = "ms_version_id" // MySensors firmware version id - LabelSmartSleepNode = "ms_smart_sleep_node" // set true, if it is a smart sleeping node + LabelImperialSystem = "ms_imperial_system" // this is bool label, used to configure about the system metric or imperial + LabelNodeID = "ms_node_id" // MySensors node id + LabelSensorID = "ms_sensor_id" // MySensors sensor id + LabelType = "ms_type" // MySensors type reference, can be used for sensor fields + LabelTypeString = "ms_type_string" // MySensors type in string format + LabelNodeType = "ms_node_type" // MySensors node type + LabelLockedReason = "ms_locked_reason" // If the the node is locked, reason will be in this label + LabelEraseEEPROM = "ms_erase_eeprom" // Supports only for MYSBootloader, on a reboot of node, the node eeprom will be erased + LabelFirmwareTypeID = "ms_type_id" // MySensors firmware type id + LabelFirmwareVersionID = "ms_version_id" // MySensors firmware version id + LabelSmartSleepNode = "ms_smart_sleep_node" // set true, if it is a smart sleeping node + LabelSleepCommandSensorID = "ms_sleep_command_sensor_id" // LabelFotaScript is the data_repository id that holds custom FOTA policy scripts // (data.onConfig required, data.onBlock optional). When set (non-empty after trim), @@ -34,6 +36,22 @@ const ( FieldAwakeDuration = "awake_duration" // smart sleep node awake duration FieldSleepDuration = "sleep_duration" // smart sleep node sleep duration + + // LabelBootloaderOtaReboot: when set to "true" on a node, MyController sends + // an I_REBOOT right before serving a firmware update. Mimics MYSController + // and is required for MYSBootloader-based nodes. + // + // Values: + // "true" -> auto-reboot (recommended for MYSBootloader nodes) + // "false" -> no automatic reboot + // absent -> same as "false" + // + // Sleeping nodes (smart_sleep + not seen recently) are queued: the reboot is + // sent when the node next reports itself awake. + LabelBootloaderOtaReboot = "bootloader_ota_reboot" + + sleepPurgeJobName = "mysensors_sleep_command_store" + sleepPurgeJobCron = "0 */5 * * * *" // every 5 minutes, same as firmware purge ) // internal references @@ -269,10 +287,9 @@ var customValidActions = []string{ "I_CONFIG", "I_ID_REQUEST", "I_TIME", - - // stream message type requests "ST_FIRMWARE_CONFIG_REQUEST", "ST_FIRMWARE_REQUEST", + nodeTY.ActionSetSleepDuration, } // this struct used to construct payload metric type and unit diff --git a/plugin/gateway/provider/mysensors_v2/msg_parser.go b/plugin/gateway/provider/mysensors_v2/msg_parser.go index fb46c70b..5690e83f 100644 --- a/plugin/gateway/provider/mysensors_v2/msg_parser.go +++ b/plugin/gateway/provider/mysensors_v2/msg_parser.go @@ -89,7 +89,10 @@ func (p *Provider) toRawMessage(msg *msgTY.Message) (*msgTY.RawMessage, error) { } if msMsg.Type == "" { switch payload.Key { - case nodeTY.ActionFirmwareUpdate, "ST_FIRMWARE_CONFIG_REQUEST", "ST_FIRMWARE_REQUEST": + case nodeTY.ActionFirmwareUpdate, + "ST_FIRMWARE_CONFIG_REQUEST", + "ST_FIRMWARE_REQUEST", + nodeTY.ActionSetSleepDuration: p.logger.Debug("firmware action produced no message", zap.String("action", payload.Key)) return nil, nil } @@ -241,7 +244,7 @@ func (p *Provider) ConvertToMessages(rawMsg *msgTY.RawMessage) ([]*msgTY.Message // else: return? case cmdInternal: - proceedFurther, extraMessages, err := updateNodeInternalMessages(msg, &msgPL, msMsg) + proceedFurther, extraMessages, err := p.updateNodeInternalMessages(msg, &msgPL, msMsg) if err != nil || !proceedFurther { return nil, err } @@ -378,7 +381,7 @@ func (p *Provider) updateFieldAndUnit(msMsg *message, msgPL *msgTY.Payload) erro // updates node internal message data, return true, if further actions required // returns false, if the message can be dropped -func updateNodeInternalMessages(msg *msgTY.Message, msgPL *msgTY.Payload, msMsg *message) (bool, []*msgTY.Message, error) { +func (p *Provider) updateNodeInternalMessages(msg *msgTY.Message, msgPL *msgTY.Payload, msMsg *message) (bool, []*msgTY.Message, error) { if msMsg.Ack == "1" { // do not care about internal ack messages return false, nil, nil } @@ -436,10 +439,19 @@ func updateNodeInternalMessages(msg *msgTY.Message, msgPL *msgTY.Payload, msMsg } extraMessages = append(extraMessages, awakeActionMsg) + // Deliver any pending reboot, then any pending sleep command. + // Sequential to guarantee the reboot lands before the new + // sleep duration is pushed to the node. + go func(gwID, nodeID string) { + p.deliverQueuedRebootIfAny(gwID, nodeID) + p.deliverQueuedSleepCommandIfAny(gwID, nodeID) + }(msg.GatewayID, msMsg.NodeID) + case "I_POST_SLEEP_NOTIFICATION": msgPL.Others.Set(FieldSleepDuration, msMsg.Payload, nil) } + default: // noop } diff --git a/plugin/gateway/provider/mysensors_v2/ota_impl.go b/plugin/gateway/provider/mysensors_v2/ota_impl.go index 0d63c24f..c7a65767 100644 --- a/plugin/gateway/provider/mysensors_v2/ota_impl.go +++ b/plugin/gateway/provider/mysensors_v2/ota_impl.go @@ -40,6 +40,11 @@ func (p *Provider) executeFirmwareConfigRequest(msg *msgTY.Message) (string, err return "", err } + // Detect and persist bootloader kind + if rxPL != "" { + p.rememberBootloaderKind(node, rxPL) + } + fromReq := otaBlockSizeFromRequest(rxPL) if fromReq > 0 { p.rememberOtaBlockSize(node, fromReq) diff --git a/plugin/gateway/provider/mysensors_v2/ota_reboot.go b/plugin/gateway/provider/mysensors_v2/ota_reboot.go new file mode 100644 index 00000000..d07ef653 --- /dev/null +++ b/plugin/gateway/provider/mysensors_v2/ota_reboot.go @@ -0,0 +1,166 @@ +package mysensors + +import ( + "fmt" + "time" + + msgTY "github.com/mycontroller-org/server/v2/pkg/types/message" + nodeTY "github.com/mycontroller-org/server/v2/pkg/types/node" + "go.uber.org/zap" +) + +// smartSleepRecentlySeen is the window during which a smart-sleep node is +// considered "awake". Below this threshold, we assume it can receive messages. +const smartSleepRecentlySeen = 30 * time.Second + +// shouldRebootForOta reports whether the node opts into automatic reboot for OTA. +// The label must be "true" (case-insensitive). Missing or "false" means no. +func shouldRebootForOta(node *nodeTY.Node) bool { + if node == nil || node.Labels == nil { + return false + } + return node.Labels.GetBool(LabelBootloaderOtaReboot) +} + +// isSleepingNode reports whether the node uses smart sleep. +func isSleepingNode(node *nodeTY.Node) bool { + if node == nil || node.Labels == nil { + return false + } + return node.Labels.GetBool(LabelSmartSleepNode) +} + +// isRecentlySeen reports whether the node was seen within the given window. +func isRecentlySeen(node *nodeTY.Node, threshold time.Duration) bool { + if node == nil { + return false + } + return time.Since(node.LastSeen) < threshold +} + +// maxSleepDuration returns an upper bound on the node's sleep cycle. +// Falls back to 5 minutes when the label is missing. +func maxSleepDuration(node *nodeTY.Node) time.Duration { + if node == nil || node.Labels == nil { + return 5 * time.Minute + } + ms := int64(node.Labels.GetInt(FieldSleepDuration)) + if ms <= 0 { + return 5 * time.Minute + } + d := time.Duration(ms) * time.Millisecond + if d < time.Minute { + d = time.Minute + } + if d > time.Hour { + d = time.Hour + } + return d +} + +// maybeSendRebootForOta is called before serving a firmware-update action. +// It sends I_REBOOT to awake nodes, or queues it for sleeping ones. +func (p *Provider) maybeSendRebootForOta(msg *msgTY.Message) error { + node, err := p.getNode(msg.GatewayID, msg.NodeID) + if err != nil { + return fmt.Errorf("load node for OTA reboot check: %w", err) + } + if !shouldRebootForOta(node) { + p.logger.Debug("bootloader_ota_reboot disabled, skipping reboot", + zap.String("nodeId", node.NodeID), + zap.String("gatewayId", node.GatewayID), + ) + return nil + } + + // Case 1: sleeping node (smart sleep) not seen recently -> queue the reboot + if isSleepingNode(node) && !isRecentlySeen(node, smartSleepRecentlySeen) { + p.queueRebootForNode(node) + p.logger.Info("node is sleeping, I_REBOOT queued until next wake", + zap.String("nodeId", node.NodeID), + zap.String("gatewayId", node.GatewayID), + zap.String("sleepDuration", node.Labels.Get(FieldSleepDuration)), + zap.Duration("maxWait", maxSleepDuration(node)), + ) + return nil + } + + // Case 2: awake node, or non smart-sleep node -> send immediately + return p.sendRebootNow(node) +} + +// sendRebootNow sends an I_REBOOT to the node immediately. +func (p *Provider) sendRebootNow(node *nodeTY.Node) error { + rebootMsg := msgTY.NewMessage(false) + rebootMsg.GatewayID = node.GatewayID + rebootMsg.NodeID = node.NodeID + rebootMsg.Type = msgTY.TypeAction + + pl := msgTY.NewPayload() + pl.Key = nodeTY.ActionReboot + rebootMsg.Payloads = []msgTY.Payload{pl} + + if err := p.Post(&rebootMsg); err != nil { + return fmt.Errorf("post reboot for OTA: %w", err) + } + + p.logger.Info("sent I_REBOOT before OTA", + zap.String("nodeId", node.NodeID), + zap.String("gatewayId", node.GatewayID), + zap.String("bootloaderKind", node.Labels.Get(LabelBootloaderKind)), + zap.String("bootloaderVersion", node.Labels.Get(LabelBootloaderVersion)), + ) + return nil +} + +// queueRebootForNode stores a pending I_REBOOT for delivery when the node wakes. +func (p *Provider) queueRebootForNode(node *nodeTY.Node) { + if node == nil { + return + } + key := p.getNodeStoreID(node.GatewayID, node.NodeID) + otaPendingRebootStore.Add(key, time.Now()) +} + +// deliverQueuedRebootIfAny sends a pending I_REBOOT when the node is known to be awake. +// Safe to call multiple times; the entry is removed on first successful delivery. +func (p *Provider) deliverQueuedRebootIfAny(gatewayID, nodeID string) { + key := p.getNodeStoreID(gatewayID, nodeID) + v := otaPendingRebootStore.Get(key) + if v == nil { + return + } + otaPendingRebootStore.Remove(key) + + node, err := p.getNode(gatewayID, nodeID) + if err != nil { + p.logger.Error("failed to load node for queued reboot delivery", + zap.String("gatewayId", gatewayID), + zap.String("nodeId", nodeID), + zap.Error(err), + ) + return + } + + p.logger.Info("delivering queued I_REBOOT to awake node", + zap.String("nodeId", node.NodeID), + zap.String("gatewayId", node.GatewayID), + ) + + // Allow a short delay so the node has completed its wake-up sequence + // (heartbeat, I_CONFIG, presentation, ...) before receiving I_REBOOT. + go func() { + time.Sleep(500 * time.Millisecond) + if err := p.sendRebootNow(node); err != nil { + p.logger.Error("failed to deliver queued reboot", + zap.String("nodeId", node.NodeID), + zap.Error(err), + ) + } + }() +} + +// pendingRebootCount returns the number of queued reboots (used by tests). +func pendingRebootCount() int { + return len(otaPendingRebootStore.Keys()) +} diff --git a/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go b/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go new file mode 100644 index 00000000..71b2b403 --- /dev/null +++ b/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go @@ -0,0 +1,85 @@ +package mysensors + +import ( + "testing" + "time" + + "github.com/mycontroller-org/server/v2/pkg/types/cmap" + nodeTY "github.com/mycontroller-org/server/v2/pkg/types/node" +) + +func TestShouldRebootForOta(t *testing.T) { + tests := []struct { + name string + label string + want bool + }{ + {"absent", "", false}, + {"true", "true", true}, + {"false", "false", false}, + {"TRUE", "TRUE", true}, + {"1", "1", true}, + {"0", "0", false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + labels := cmap.CustomStringMap{} + if tt.label != "" { + labels.Set(LabelBootloaderOtaReboot, tt.label) + } + n := &nodeTY.Node{Labels: labels} + if got := shouldRebootForOta(n); got != tt.want { + t.Errorf("shouldRebootForOta(%q) = %v, want %v", tt.label, got, tt.want) + } + }) + } + + if shouldRebootForOta(nil) { + t.Fatal("nil node must be false") + } +} + +func TestIsSleepingNode(t *testing.T) { + n := &nodeTY.Node{Labels: cmap.CustomStringMap{}} + if isSleepingNode(n) { + t.Fatal("empty labels must not be sleeping") + } + n.Labels.Set(LabelSmartSleepNode, "true") + if !isSleepingNode(n) { + t.Fatal("smart_sleep_node=true must be sleeping") + } + n.Labels.Set(LabelSmartSleepNode, "false") + if isSleepingNode(n) { + t.Fatal("smart_sleep_node=false must not be sleeping") + } +} + +func TestIsRecentlySeen(t *testing.T) { + n := &nodeTY.Node{LastSeen: time.Now()} + if !isRecentlySeen(n, 30*time.Second) { + t.Fatal("just seen must be recent") + } + n.LastSeen = time.Now().Add(-1 * time.Minute) + if isRecentlySeen(n, 30*time.Second) { + t.Fatal("1 min ago must not be recent for 30s threshold") + } +} + +func TestMaxSleepDuration(t *testing.T) { + n := &nodeTY.Node{Labels: cmap.CustomStringMap{}} + if got := maxSleepDuration(n); got != 5*time.Minute { + t.Fatalf("default: %v", got) + } + n.Labels.Set(FieldSleepDuration, "280000") // 280000 ms = 280 s + if got := maxSleepDuration(n); got < 4*time.Minute || got > 5*time.Minute { + t.Fatalf("280000ms: %v", got) + } + n.Labels.Set(FieldSleepDuration, "10000") // 10 s -> clamped to 1 min + if got := maxSleepDuration(n); got != time.Minute { + t.Fatalf("10s clamp: %v", got) + } + n.Labels.Set(FieldSleepDuration, "7200000") // 2 h -> clamped to 1 h + if got := maxSleepDuration(n); got != time.Hour { + t.Fatalf("2h clamp: %v", got) + } +} diff --git a/plugin/gateway/provider/mysensors_v2/ota_store.go b/plugin/gateway/provider/mysensors_v2/ota_store.go index 5d723fa9..69a08d98 100644 --- a/plugin/gateway/provider/mysensors_v2/ota_store.go +++ b/plugin/gateway/provider/mysensors_v2/ota_store.go @@ -15,13 +15,24 @@ import ( ) var ( - nodeStore = concurrency.NewStore() - fwStore = concurrency.NewStore() - fwRawStore = concurrency.NewStore() - fotaScriptStore = concurrency.NewStore() // data_repository id → *fotaScriptBundle - fotaSessionStore = concurrency.NewStore() // gateway_node → *fotaSession + nodeStore = concurrency.NewStore() + fwStore = concurrency.NewStore() + fwRawStore = concurrency.NewStore() + fotaScriptStore = concurrency.NewStore() // data_repository id -> *fotaScriptBundle + fotaSessionStore = concurrency.NewStore() // gateway_node -> *fotaSession + otaPendingRebootStore = concurrency.NewStore() // gateway_node -> time.Time + pendingSleepStore = concurrency.NewStore() // gateway_node -> pendingSleepCommand ) +// pendingSleepCommand is a sleep-duration change command queued for a sleeping +// smart_sleep node. +type pendingSleepCommand struct { + SensorID string // child sensor id that receives the V_PERCENTAGE + Percent uint8 // V_PERCENTAGE value to send (0-100) + QueuedAt time.Time // for TTL / purge + RequestID string // optional: correlation with the caller +} + // fotaSession is the firmware + slice size advertised in onConfig, reused for every block. type fotaSession struct { FirmwareID string @@ -68,7 +79,7 @@ func (p *Provider) getNode(gatewayID, nodeID string) (*nodeTY.Node, error) { return nil, fmt.Errorf("node not available. gatewayID:%s, nodeID:%s", gatewayID, nodeID) } -// getNode returns the node +// getFirmware returns the firmware func (p *Provider) getFirmware(id string) (*firmwareTY.Firmware, error) { toFirmware := func(item interface{}) (*firmwareTY.Firmware, error) { if fw, ok := item.(*firmwareTY.Firmware); ok { diff --git a/plugin/gateway/provider/mysensors_v2/provider.go b/plugin/gateway/provider/mysensors_v2/provider.go index 13f46bf7..ebc01193 100644 --- a/plugin/gateway/provider/mysensors_v2/provider.go +++ b/plugin/gateway/provider/mysensors_v2/provider.go @@ -113,19 +113,26 @@ func (p *Provider) Start(receivedMessageHandler func(rawMsg *msgTY.RawMessage) e protocol, _err := ethernet.New(p.logger, p.GatewayConfig, p.Config.Protocol, receivedMessageHandler, p.bus, p.logRootDirectory) err = _err p.Protocol = protocol + } + if err != nil { + return err } + // load firmware purge job (per gateway, suffix with gateway id) + fwPurgeJobName := fmt.Sprintf("%s_%s", firmwarePurgeJobName, p.GatewayConfig.ID) + err = p.scheduler.AddFunc(fwPurgeJobName, firmwarePurgeJobCron, firmwareRawPurge) if err != nil { return err } - // load firmware purge job - firmwarePurgeJobName := fmt.Sprintf("%s_%s", firmwarePurgeJobName, p.GatewayConfig.ID) - err = p.scheduler.AddFunc(firmwarePurgeJobName, firmwarePurgeJobCron, firmwareRawPurge) + // load sleep command purge job (per gateway, suffix with gateway id) + sleepJobName := fmt.Sprintf("%s_%s", sleepPurgeJobName, p.GatewayConfig.ID) + err = p.scheduler.AddFunc(sleepJobName, sleepPurgeJobCron, purgeExpiredSleepCommands) if err != nil { return err } + err = p.initEventListener(p.GatewayConfig.ID) if err != nil { return err @@ -145,6 +152,11 @@ func (p *Provider) Close() error { // remove firmware purge job fwPurgeJobName := fmt.Sprintf("%s_%s", firmwarePurgeJobName, p.GatewayConfig.ID) p.scheduler.RemoveFunc(fwPurgeJobName) + + // remove sleep command purge job + sleepJobName := fmt.Sprintf("%s_%s", sleepPurgeJobName, p.GatewayConfig.ID) + p.scheduler.RemoveFunc(sleepJobName) + // close gateway return p.Protocol.Close() } diff --git a/plugin/gateway/provider/mysensors_v2/sleep_command.go b/plugin/gateway/provider/mysensors_v2/sleep_command.go new file mode 100644 index 00000000..2de2ffde --- /dev/null +++ b/plugin/gateway/provider/mysensors_v2/sleep_command.go @@ -0,0 +1,204 @@ +package mysensors + +import ( + "fmt" + "strconv" + "strings" + "time" + + msgTY "github.com/mycontroller-org/server/v2/pkg/types/message" + nodeTY "github.com/mycontroller-org/server/v2/pkg/types/node" + "go.uber.org/zap" +) + +// sleepRecentlySeen is the same threshold as the OTA reboot logic. +// A node seen within this window is considered awake. +const sleepRecentlySeen = 30 * time.Second + +// sleepCommandVPercentage is the numeric code for V_PERCENTAGE +// (see setReqFieldMapForRx in constant.go). +const sleepCommandVPercentage = "3" + +// sleepCommandMaxAge is the TTL of a queued command. +// Beyond this, the command is considered stale and purged. +const sleepCommandMaxAge = 2 * time.Hour + +// parseSleepDurationPayload extracts the optional sensor_id and the percentage. +// +// Accepted formats: +// +// "32" -> ("", 32, nil) +// "206:32" -> ("206", 32, nil) +// " 206 : 32 " -> ("206", 32, nil) +// "foo" -> ("", 0, err) +// "206:200" -> ("", 0, err) +// "206:foo" -> ("", 0, err) +func parseSleepDurationPayload(raw string) (sensorID string, percent uint8, err error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return "", 0, fmt.Errorf("empty payload") + } + valueStr := raw + if idx := strings.Index(raw, ":"); idx >= 0 { + sensorID = strings.TrimSpace(raw[:idx]) + valueStr = strings.TrimSpace(raw[idx+1:]) + } + pct, convErr := strconv.Atoi(valueStr) + if convErr != nil || pct < 0 || pct > 100 { + return "", 0, fmt.Errorf("invalid sleep percentage: %q (expected integer 0-100)", valueStr) + } + return sensorID, uint8(pct), nil +} + +// sendSleepCommand sends (or queues) a sleep-duration change to a smart_sleep +// node. +// +// If the node is awake (recent contact), the command goes out immediately. +// Otherwise it is stored and delivered on the next I_PRE_SLEEP_NOTIFICATION +// or I_HEARTBEAT_RESPONSE (see deliverQueuedSleepCommandIfAny). +func (p *Provider) sendSleepCommand(node *nodeTY.Node, sensorID string, percent uint8, requestID string) error { + if node == nil { + return fmt.Errorf("nil node") + } + if sensorID == "" { + return fmt.Errorf("sleep command sensor id is empty") + } + if percent > 100 { + return fmt.Errorf("invalid sleep percentage: %d", percent) + } + if !isSleepingNode(node) { + return fmt.Errorf("node %s is not a smart_sleep node", node.ID) + } + + if isRecentlySeen(node, sleepRecentlySeen) { + return p.sendSleepCommandNow(node, sensorID, percent) + } + + p.queueSleepCommand(node, sensorID, percent, requestID) + p.logger.Info("sleep command queued until next wake", + zap.String("nodeId", node.NodeID), + zap.String("gatewayId", node.GatewayID), + zap.String("sensorId", sensorID), + zap.Uint8("percent", percent), + zap.String("requestId", requestID), + zap.String("sleepDuration", node.Labels.Get(FieldSleepDuration)), + zap.Duration("maxWait", maxSleepDuration(node)), + ) + return nil +} + +// sendSleepCommandNow sends the C_SET/V_PERCENTAGE to the node immediately. +func (p *Provider) sendSleepCommandNow(node *nodeTY.Node, sensorID string, percent uint8) error { + sleepMsg := msgTY.NewMessage(false) + sleepMsg.GatewayID = node.GatewayID + sleepMsg.NodeID = node.NodeID + sleepMsg.SourceID = sensorID + sleepMsg.Type = msgTY.TypeSet + + pl := msgTY.NewPayload() + pl.Key = "V_PERCENTAGE" + pl.SetValue(fmt.Sprintf("%d", percent)) + pl.Labels.Set(LabelType, sleepCommandVPercentage) + + sleepMsg.Payloads = []msgTY.Payload{pl} + + if err := p.Post(&sleepMsg); err != nil { + return fmt.Errorf("post sleep command: %w", err) + } + + p.logger.Info("sent sleep command", + zap.String("nodeId", node.NodeID), + zap.String("gatewayId", node.GatewayID), + zap.String("sensorId", sensorID), + zap.Uint8("percent", percent), + ) + return nil +} + +// queueSleepCommand stores the command for later delivery. +func (p *Provider) queueSleepCommand(node *nodeTY.Node, sensorID string, percent uint8, requestID string) { + if node == nil { + return + } + key := p.getNodeStoreID(node.GatewayID, node.NodeID) + pendingSleepStore.Add(key, pendingSleepCommand{ + SensorID: sensorID, + Percent: percent, + QueuedAt: time.Now(), + RequestID: requestID, + }) +} + +// deliverQueuedSleepCommandIfAny sends the queued command when the node wakes. +// Called from updateNodeInternalMessages() on I_PRE_SLEEP_NOTIFICATION / +// I_HEARTBEAT_RESPONSE, same as the OTA reboot delivery. +// +// Idempotent: the entry is removed before sending to avoid duplicates. +func (p *Provider) deliverQueuedSleepCommandIfAny(gatewayID, nodeID string) { + key := p.getNodeStoreID(gatewayID, nodeID) + v := pendingSleepStore.Get(key) + if v == nil { + return + } + cmd, ok := v.(pendingSleepCommand) + if !ok { + pendingSleepStore.Remove(key) + return + } + if time.Since(cmd.QueuedAt) > sleepCommandMaxAge { + pendingSleepStore.Remove(key) + p.logger.Warn("pending sleep command expired", + zap.String("gatewayId", gatewayID), + zap.String("nodeId", nodeID), + ) + return + } + pendingSleepStore.Remove(key) + + node, err := p.getNode(gatewayID, nodeID) + if err != nil { + p.logger.Error("failed to load node for queued sleep command", + zap.String("gatewayId", gatewayID), + zap.String("nodeId", nodeID), + zap.Error(err), + ) + return + } + + p.logger.Info("delivering queued sleep command to awake node", + zap.String("nodeId", node.NodeID), + zap.String("gatewayId", node.GatewayID), + zap.String("sensorId", cmd.SensorID), + zap.Uint8("percent", cmd.Percent), + ) + + // Allow a short delay so the node has completed its wake-up sequence + // (heartbeat, I_CONFIG, presentation, ...) before receiving the C_SET. + go func() { + time.Sleep(500 * time.Millisecond) + if err := p.sendSleepCommandNow(node, cmd.SensorID, cmd.Percent); err != nil { + p.logger.Error("failed to deliver queued sleep command", + zap.String("nodeId", node.NodeID), + zap.Error(err), + ) + } + }() +} + +// pendingSleepCount is used by tests. +func pendingSleepCount() int { + return len(pendingSleepStore.Keys()) +} + +// purgeExpiredSleepCommands is scheduled like firmwareRawPurge(). +// Removes commands that are too old. +func purgeExpiredSleepCommands() { + now := time.Now() + for _, key := range pendingSleepStore.Keys() { + v := pendingSleepStore.Get(key) + cmd, ok := v.(pendingSleepCommand) + if !ok || now.Sub(cmd.QueuedAt) > sleepCommandMaxAge { + pendingSleepStore.Remove(key) + } + } +} diff --git a/plugin/gateway/provider/mysensors_v2/sleep_command_test.go b/plugin/gateway/provider/mysensors_v2/sleep_command_test.go new file mode 100644 index 00000000..481eeb01 --- /dev/null +++ b/plugin/gateway/provider/mysensors_v2/sleep_command_test.go @@ -0,0 +1,123 @@ +package mysensors + +import ( + "testing" + "time" + + "github.com/mycontroller-org/server/v2/pkg/types/cmap" + nodeTY "github.com/mycontroller-org/server/v2/pkg/types/node" +) + +// resetPendingSleepStore clears the store to isolate tests. +// concurrency.Store does not expose Reset(), so we iterate over the keys. +func resetPendingSleepStore() { + for _, k := range pendingSleepStore.Keys() { + pendingSleepStore.Remove(k) + } +} + +func TestSleepCommandIsSleepingNode(t *testing.T) { + n := &nodeTY.Node{Labels: cmap.CustomStringMap{}} + if isSleepingNode(n) { + t.Fatal("empty labels must not be sleeping") + } + n.Labels.Set(LabelSmartSleepNode, "true") + if !isSleepingNode(n) { + t.Fatal("smart_sleep_node=true must be sleeping") + } +} + +func TestSleepCommandRecentlySeen(t *testing.T) { + n := &nodeTY.Node{LastSeen: time.Now()} + if !isRecentlySeen(n, sleepRecentlySeen) { + t.Fatal("just seen must be recent") + } + n.LastSeen = time.Now().Add(-1 * time.Minute) + if isRecentlySeen(n, sleepRecentlySeen) { + t.Fatal("1 min ago must not be recent for 30s threshold") + } +} + +func TestPendingSleepCount(t *testing.T) { + resetPendingSleepStore() + if pendingSleepCount() != 0 { + t.Fatal("must start empty") + } + node := &nodeTY.Node{GatewayID: "gw", NodeID: "1"} + p := &Provider{} + p.queueSleepCommand(node, "206", 50, "req-1") + if pendingSleepCount() != 1 { + t.Fatalf("want 1 got %d", pendingSleepCount()) + } +} + +func TestPurgeExpiredSleepCommands(t *testing.T) { + resetPendingSleepStore() + key := "gw_1" + pendingSleepStore.Add(key, pendingSleepCommand{ + SensorID: "206", + Percent: 75, + QueuedAt: time.Now().Add(-3 * time.Hour), + }) + purgeExpiredSleepCommands() + if pendingSleepCount() != 0 { + t.Fatalf("expired command must be purged") + } +} + +// --- Additional tests --- + +func TestParseSleepDurationPayload(t *testing.T) { + tests := []struct { + in string + wantID string + wantPct uint8 + wantErr bool + }{ + {"32", "", 32, false}, + {"0", "", 0, false}, + {"100", "", 100, false}, + {"206:32", "206", 32, false}, + {" 206 : 32 ", "206", 32, false}, + {"206:100", "206", 100, false}, + {"", "", 0, true}, + {"101", "", 0, true}, + {"-1", "", 0, true}, + {"foo", "", 0, true}, + {"206:foo", "", 0, true}, + {"206:200", "", 0, true}, + } + for _, tt := range tests { + t.Run(tt.in, func(t *testing.T) { + id, pct, err := parseSleepDurationPayload(tt.in) + if tt.wantErr { + if err == nil { + t.Fatalf("expected error for %q", tt.in) + } + return + } + if err != nil { + t.Fatalf("unexpected error for %q: %v", tt.in, err) + } + if id != tt.wantID || pct != tt.wantPct { + t.Fatalf("got (%q,%d), want (%q,%d)", id, pct, tt.wantID, tt.wantPct) + } + }) + } +} + +func TestPendingSleepCommandKeepsSensorID(t *testing.T) { + resetPendingSleepStore() + node := &nodeTY.Node{GatewayID: "gw", NodeID: "1"} + p := &Provider{} + p.queueSleepCommand(node, "206", 50, "req-1") + + v := pendingSleepStore.Get("gw_1") + cmd, ok := v.(pendingSleepCommand) + if !ok { + t.Fatal("expected pendingSleepCommand") + } + if cmd.SensorID != "206" || cmd.Percent != 50 { + t.Fatalf("got SensorID=%q Percent=%d", cmd.SensorID, cmd.Percent) + } +} From ca74c8230062d70946faba74a51a38c20405d383 Mon Sep 17 00:00:00 2001 From: Eddine97 Date: Sat, 17 Oct 2026 02:03:56 +0200 Subject: [PATCH 3/4] feat(mysensors): trigger OTA reboot automatically for MYSBootloader nodes --- .../provider/mysensors_v2/ota_reboot.go | 23 +++++++-- .../provider/mysensors_v2/ota_reboot_test.go | 51 +++++++++++++++++++ 2 files changed, 70 insertions(+), 4 deletions(-) diff --git a/plugin/gateway/provider/mysensors_v2/ota_reboot.go b/plugin/gateway/provider/mysensors_v2/ota_reboot.go index d07ef653..f5ce4b43 100644 --- a/plugin/gateway/provider/mysensors_v2/ota_reboot.go +++ b/plugin/gateway/provider/mysensors_v2/ota_reboot.go @@ -60,32 +60,47 @@ func maxSleepDuration(node *nodeTY.Node) time.Duration { // maybeSendRebootForOta is called before serving a firmware-update action. // It sends I_REBOOT to awake nodes, or queues it for sleeping ones. +// +// A reboot is needed when either: +// - the user explicitly opted in with bootloader_ota_reboot=true, or +// - the detected bootloader is a MYSBootloader variant (any 1.x). +// +// Protocol 3.1 / DualOptiboot nodes do not need a reboot: they enter OTA on +// their own when they receive the firmware config request. func (p *Provider) maybeSendRebootForOta(msg *msgTY.Message) error { node, err := p.getNode(msg.GatewayID, msg.NodeID) if err != nil { return fmt.Errorf("load node for OTA reboot check: %w", err) } - if !shouldRebootForOta(node) { - p.logger.Debug("bootloader_ota_reboot disabled, skipping reboot", + + explicitOptIn := shouldRebootForOta(node) + detectedMYSBootloader := isMYSBootloaderKind(bootloaderKindForNode(node)) + + if !explicitOptIn && !detectedMYSBootloader { + p.logger.Debug("no pre-OTA reboot needed", zap.String("nodeId", node.NodeID), zap.String("gatewayId", node.GatewayID), + zap.String("bootloaderKind", bootloaderKindForNode(node)), + zap.Bool("explicitOptIn", explicitOptIn), ) return nil } - // Case 1: sleeping node (smart sleep) not seen recently -> queue the reboot + // A reboot is needed. Queue it if the node is a smart-sleep node that is + // currently asleep; otherwise send it immediately. if isSleepingNode(node) && !isRecentlySeen(node, smartSleepRecentlySeen) { p.queueRebootForNode(node) p.logger.Info("node is sleeping, I_REBOOT queued until next wake", zap.String("nodeId", node.NodeID), zap.String("gatewayId", node.GatewayID), + zap.String("bootloaderKind", bootloaderKindForNode(node)), + zap.Bool("explicitOptIn", explicitOptIn), zap.String("sleepDuration", node.Labels.Get(FieldSleepDuration)), zap.Duration("maxWait", maxSleepDuration(node)), ) return nil } - // Case 2: awake node, or non smart-sleep node -> send immediately return p.sendRebootNow(node) } diff --git a/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go b/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go index 71b2b403..f6bf10db 100644 --- a/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go +++ b/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go @@ -83,3 +83,54 @@ func TestMaxSleepDuration(t *testing.T) { t.Fatalf("2h clamp: %v", got) } } + +func TestMaybeSendRebootDecision(t *testing.T) { + tests := []struct { + name string + labels cmap.CustomStringMap + wantNeedsBoot bool + }{ + { + name: "no labels, no detection", + labels: cmap.CustomStringMap{}, + wantNeedsBoot: false, + }, + { + name: "explicit opt-in", + labels: cmap.CustomStringMap{LabelBootloaderOtaReboot: "true"}, + wantNeedsBoot: true, + }, + { + name: "detected MYSBootloader 1.4", + labels: cmap.CustomStringMap{LabelBootloaderKind: BootloaderKindMYSBootloader14}, + wantNeedsBoot: true, + }, + { + name: "detected MYSBootloader 1.0", + labels: cmap.CustomStringMap{LabelBootloaderKind: BootloaderKindMYSBootloader10}, + wantNeedsBoot: true, + }, + { + name: "detected protocol 3.1, no opt-in", + labels: cmap.CustomStringMap{LabelBootloaderKind: BootloaderKindProtocol31}, + wantNeedsBoot: false, + }, + { + name: "protocol 3.1 with explicit opt-in", + labels: cmap.CustomStringMap{ + LabelBootloaderKind: BootloaderKindProtocol31, + LabelBootloaderOtaReboot: "true", + }, + wantNeedsBoot: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + n := &nodeTY.Node{Labels: tt.labels} + got := shouldRebootForOta(n) || isMYSBootloaderKind(bootloaderKindForNode(n)) + if got != tt.wantNeedsBoot { + t.Errorf("needsReboot = %v, want %v", got, tt.wantNeedsBoot) + } + }) + } +} From c96030109d498f4b95fb9b395fd12e3903ebb8a7 Mon Sep 17 00:00:00 2001 From: Eddine97 Date: Sat, 17 Oct 2026 21:52:48 +0200 Subject: [PATCH 4/4] test: cover pendingRebootCount pendingRebootCount is exposed for tests but was not exercised by any test, which golangci-lint flagged as unused. Add a minimal test that covers the empty, add, remove, and cleanup paths. --- .../provider/mysensors_v2/ota_reboot_test.go | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go b/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go index f6bf10db..74cc79d9 100644 --- a/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go +++ b/plugin/gateway/provider/mysensors_v2/ota_reboot_test.go @@ -134,3 +134,38 @@ func TestMaybeSendRebootDecision(t *testing.T) { }) } } + +func TestPendingRebootCount(t *testing.T) { + // Nettoyer l'état entre les tests (le store est global au package). + for _, k := range otaPendingRebootStore.Keys() { + otaPendingRebootStore.Remove(k) + } + defer func() { + for _, k := range otaPendingRebootStore.Keys() { + otaPendingRebootStore.Remove(k) + } + }() + + if n := pendingRebootCount(); n != 0 { + t.Fatalf("expected 0 pending reboots, got %d", n) + } + + otaPendingRebootStore.Add("gw1.node1", time.Now()) + otaPendingRebootStore.Add("gw1.node2", time.Now()) + + if n := pendingRebootCount(); n != 2 { + t.Fatalf("expected 2 pending reboots, got %d", n) + } + + otaPendingRebootStore.Remove("gw1.node1") + + if n := pendingRebootCount(); n != 1 { + t.Fatalf("expected 1 pending reboot, got %d", n) + } + + otaPendingRebootStore.Remove("gw1.node2") + + if n := pendingRebootCount(); n != 0 { + t.Fatalf("expected 0 pending reboots after cleanup, got %d", n) + } +}