Read Shelly input type through RPC

Author: MaxRink

src/devices/shelly.ts

@@ -1595,16 +1595,64 @@ const fzLocal = {
     } satisfies Fz.Converter<"ssIasZone", undefined, ["commandStatusChangeNotification", "attributeReport", "readResponse"]>,
 };
 
+const shellyRpcResult = (response: KeyValue | undefined): KeyValue | undefined => {
+    const result = response?.result ?? response?.params ?? response;
+    if (!result) return undefined;
+    assertObject<KeyValue>(result);
+    return result;
+};
+
+const getShellyInputId = (meta: Pick<Tz.Meta, "endpoint_name">): number => {
+    if (meta.endpoint_name === "sw2") return 1;
+    return 0;
+};
+
+const getShellyRpcEndpoint = (entity: Zh.Endpoint | Zh.Group): Zh.Endpoint | undefined => {
+    if (!utils.isEndpoint(entity)) return undefined;
+    return entity.getDevice().getEndpoint(SHELLY_ENDPOINT_ID);
+};
+
+const shellyInputTypeLookup = {
+    switch: "toggle",
+    button: "momentary",
+} as const;
+
+const shellyInputTypeSetLookup = {
+    toggle: "switch",
+    momentary: "button",
+} as const;
+
 const tzLocal = {
     switch_input_type: {
         key: ["switch_type"],
         convertSet: async (entity, key, value, meta) => {
             const lookup = {toggle: 0, momentary: 1} as const;
+            const rpcEndpoint = getShellyRpcEndpoint(entity);
+            if (rpcEndpoint) {
+                const inputId = getShellyInputId(meta);
+                await shellyRpcSend(rpcEndpoint, "Input.SetConfig", {
+                    id: inputId,
+                    config: {type: utils.getFromLookup(value as string, shellyInputTypeSetLookup)},
+                });
+                return {state: {switch_type: value}};
+            }
+
             const ep = determineEndpoint(entity, meta, "genOnOffSwitchCfg");
             await ep.write("genOnOffSwitchCfg", {switchType: utils.getFromLookup(value as string, lookup)});
             return {state: {switch_type: value}};
         },
         convertGet: async (entity, key, meta) => {
+            const rpcEndpoint = getShellyRpcEndpoint(entity);
+            if (rpcEndpoint) {
+                const inputId = getShellyInputId(meta);
+                const config = shellyRpcResult(await shellyRpcRequest(rpcEndpoint, "Input.GetConfig", {id: inputId}));
+                if (typeof config?.type === "string") {
+                    const switchType = utils.getFromLookup(config.type, shellyInputTypeLookup);
+                    meta.publish({[meta.endpoint_name ? `switch_type_${meta.endpoint_name}` : "switch_type"]: switchType});
+                }
+                return;
+            }
+
             const ep = determineEndpoint(entity, meta, "genOnOffSwitchCfg");
             await ep.read("genOnOffSwitchCfg", ["switchType"]);
         },

test/shelly.test.ts

@@ -97,6 +97,48 @@ describe("Shelly 2PM Gen4 cover mode", () => {
         expect(state).toStrictEqual({switch_type_sw1: "toggle"});
     });
 
+    it("reads switch input type through the Shelly RPC endpoint", async () => {
+        const response = JSON.stringify({id: 1, result: {id: 1, type: "button", enable: true, invert: false}});
+        const read = vi.fn().mockResolvedValueOnce({rxCtl: response.length}).mockResolvedValueOnce({data: response});
+        const device = mockShelly2PMCoverWithInputs(read);
+        const definition = await findByDevice(device);
+        const converter = definition.toZigbee.find((converter) => converter.key.includes("switch_type")) as Tz.Converter;
+        const publish = vi.fn();
+
+        await converter.convertGet?.(device.getEndpoint(3), "switch_type", {endpoint_name: "sw2", message: {}, publish} as never);
+
+        expect(device.getEndpoint(239).write).toHaveBeenCalledWith(
+            "shellyRPCCluster",
+            {data: expect.stringContaining("Input.GetConfig")},
+            expect.any(Object),
+        );
+        expect(device.getEndpoint(239).write).toHaveBeenCalledWith("shellyRPCCluster", {data: expect.stringContaining('"id":1')}, expect.any(Object));
+        expect(read).toHaveBeenCalledWith("shellyRPCCluster", ["rxCtl"], expect.any(Object));
+        expect(read).toHaveBeenCalledWith("shellyRPCCluster", ["data"], expect.any(Object));
+        expect(device.getEndpoint(3).read).not.toHaveBeenCalled();
+        expect(publish).toHaveBeenCalledWith({switch_type_sw2: "momentary"});
+    });
+
+    it("writes switch input type through the Shelly RPC endpoint", async () => {
+        const device = mockShelly2PMCoverWithInputs();
+        const definition = await findByDevice(device);
+        const converter = definition.toZigbee.find((converter) => converter.key.includes("switch_type")) as Tz.Converter;
+
+        await converter.convertSet?.(device.getEndpoint(3), "switch_type", "momentary", {endpoint_name: "sw2", message: {}} as never);
+
+        expect(device.getEndpoint(239).write).toHaveBeenCalledWith(
+            "shellyRPCCluster",
+            {data: expect.stringContaining("Input.SetConfig")},
+            expect.any(Object),
+        );
+        expect(device.getEndpoint(239).write).toHaveBeenCalledWith(
+            "shellyRPCCluster",
+            {data: expect.stringContaining('"type":"button"')},
+            expect.any(Object),
+        );
+        expect(device.getEndpoint(3).write).not.toHaveBeenCalled();
+    });
+
     it("reads two-channel switch mode through the Shelly RPC endpoint", async () => {
         const response = JSON.stringify({id: 1, result: {id: 1, in_mode: "detached"}});
         const read = vi.fn().mockResolvedValueOnce({rxCtl: response.length}).mockResolvedValueOnce({data: response});