feat: expose valve/subdevice run modes (duration/volume) in Home Assistant

README.md

@@ -8,7 +8,8 @@ A .NET 10 bridge that connects Ecowitt weather stations and IoT subdevices to MQ
 - Two ingestion modes per gateway: **Push** (legacy `/data/report`, works with any Ecowitt-protocol station) or **Poll** (`get_livedata_info` API, for IoT-capable gateways)
 - Automatic discovery of new sensors and subdevices
 - Bidirectional communication with subdevices (AC1100, WFC01, WFC02)
-- Home Assistant MQTT discovery (devices, sensors, switches)
+- Valve/plug run modes in Home Assistant — duration or volume runs that the **device self-closes** (no fragile HA timers), exposed as select + number + button entities, capability-gated per device
+- Home Assistant MQTT discovery (devices, sensors, switches, numbers, selects, buttons)
 - Metric/imperial unit conversion (Push mode) — Poll mode passes the gateway's configured display unit through to HA verbatim
 - Change-detection filtering to reduce MQTT noise
 

docs/architecture.md

@@ -69,6 +69,29 @@ flowchart LR
     HTTP -->|POST /parse_quick_cmd_iot| GW[Ecowitt Gateway]
 ```
 
+### Run modes (duration / volume)
+
+The valve/plug run-mode entities (`select` + `number` + `button`) feed a separate
+staging path. `MqttService` translates each entity change into a `SubdeviceRunConfig`
+message; `Dispatcher` merges it into the subdevice's `StagedRunConfig` (held in
+`DeviceStore`) and, on the Start button, validates mode applicability + value
+bounds, assembles a `SubdeviceApiCommand`, and routes it through the **same shared
+guard** as the direct path before dispatch. Capability gating (Duration universal;
+Volume only for flow-capable valves) and the unit→gateway conversion
+(minutes→seconds `val_type:0`, liters→deciliters `val_type:3`) are enforced
+controller-side so the device self-closes at the target — the direct-MQTT path
+cannot bypass them.
+
+```mermaid
+flowchart LR
+    HA[HA select / number / button] -->|MQTT cmd/#| MQTT[MqttService]
+    MQTT -->|SubdeviceRunConfig| BUS[(Message Bus)]
+    BUS --> DISP[Dispatcher]
+    DISP -->|merge -> DeviceStore staged config| DISP
+    DISP -->|on Start: guard + SubdeviceCommandDispatch| BUS
+    BUS --> HTTP[HttpPublishingService]
+```
+
 ## Per-gateway Concurrency
 
 All HTTP traffic to a given gateway — subdevice polling, livedata polling (when in Poll mode), and command dispatch — runs through `HttpPublishingService` and serializes through a `ConcurrentDictionary<string, SemaphoreSlim>` keyed by gateway IP. At most one HTTP request per gateway is in flight at any moment. Different gateways still get full parallelism. A separate 350ms `Task.Delay` between subdevice payload calls handles rate-limiting (a separate contract from concurrency).
@@ -100,8 +123,13 @@ flowchart LR
     MQTT --> MCE[mqtt-connection-event] --> DISP
     MQTT --> HAS[home-assistant-status] --> DISP
     MQTT --> SAC[subdevice-api-command] --> DISP
+    MQTT --> SRC[subdevice-run-config] --> DISP
 ```
 
+`subdevice-api-command` carries the HA switch (ON/OFF) and direct-MQTT commands.
+`subdevice-run-config` carries the run-mode entities (select/number/button) as
+partial staged-config updates plus a Start flag.
+
 ### Dispatcher to HttpPublishingService
 
 HTTP polling configuration (subdevice + livedata hosts and intervals) and subdevice command dispatch.
@@ -130,10 +158,10 @@ flowchart LR
 ASP.NET endpoint at `POST /data/report`. Receives form-encoded weather data from Push-mode gateways and publishes `GatewayApiData` onto the bus. For Poll-mode gateways, the request is accepted (200 OK) but the payload is dropped by Dispatcher at the bus boundary — no DataController logic needs to know about IngestMode.
 
 ### Dispatcher
-Central orchestrator. Consumes messages from both HTTP and MQTT services, manages the `DeviceStore` (thread-safe in-memory state), performs change detection via `DeNoiserHelper`, and emits data and discovery messages. For Poll-mode gateways it also drops any incoming `GatewayApiData` push payload (the IngestMode filter lives here, not in DataController). Split across partial classes for HTTP consumers, MQTT consumers, and orchestration logic.
+Central orchestrator. Consumes messages from both HTTP and MQTT services, manages the `DeviceStore` (thread-safe in-memory state), performs change detection via `DeNoiserHelper`, and emits data and discovery messages. For Poll-mode gateways it also drops any incoming `GatewayApiData` push payload (the IngestMode filter lives here, not in DataController). Owns run-mode policy: stages `SubdeviceRunConfig` updates onto the subdevice, validates mode applicability + value bounds in a shared guard (so the direct-MQTT path can't bypass the HA-side constraints), and fires the assembled command on Start. Split across partial classes for HTTP consumers, MQTT consumers, and orchestration logic.
 
 ### MqttService
-Connects to the MQTT broker, publishes sensor data and Home Assistant discovery payloads, subscribes to HA status topic for re-emission on HA restart, and forwards subdevice commands from HA switches onto the bus as `SubdeviceApiCommand` messages. Split across partial classes for consumer, publisher, discovery, and events.
+Connects to the MQTT broker, publishes sensor data and Home Assistant discovery payloads (including the run-mode `select`/`number`/`button` entities for controllable subdevices), subscribes to HA status topic for re-emission on HA restart, and forwards subdevice commands onto the bus — HA switch / direct JSON as `SubdeviceApiCommand`, and run-mode entity changes as `SubdeviceRunConfig`. A single `<base>/+/subdevices/+/cmd/#` subscription covers all subdevice command topics. Split across partial classes for consumer, publisher, discovery, and events.
 
 ### HttpPublishingService
 Owns all outbound HTTP traffic to gateways. Runs two `PeriodicTimer` loops in parallel — one for subdevice polling (`get_iot_device_list` + `parse_quick_cmd_iot`), one for livedata polling (`get_livedata_info`) on Poll-mode gateways — plus a `SubdeviceCommandDispatch` bus consumer that issues start/stop commands. All three call sites acquire a per-gateway `SemaphoreSlim` to serialize requests to the same device.

docs/mqtt.md

@@ -63,13 +63,26 @@ Same payload format as gateway sensors.
 
 ### Subdevice Commands (direct MQTT)
 - **`<base>/+/subdevices/+/cmd`** — Accepts JSON commands for subdevice control.
-  ```json
-  { "cmd": 0, "id": 12345, "duration": 20, "unit": 1, "alwaysOn": false }
+  ```jsonc
+  { "cmd": "Start", "id": 12345, "duration": 20, "unit": "Minutes" }
+  // cmd: Start | Stop. unit: Seconds | Minutes | Hours | Liters (or 0|1|2|3).
+  // duration is in the chosen unit; the controller normalizes time to seconds
+  // and volume to deciliters for the gateway. Omit duration/unit for an
+  // always-on Start.
   ```
-  `cmd`: `0` = Start, `1` = Stop. `unit`: `0` = Seconds, `1` = Minutes, `2` = Hours, `3` = Liters.
 
 ### Subdevice Commands (Home Assistant)
-- **`<base>/+/subdevices/+/cmd/homeassistant`** — Simplified command topic for HA switches. Payload is plain text `ON` or `OFF`.
+- **`<base>/+/subdevices/+/cmd/homeassistant`** — Simplified command topic for HA switches. Payload is plain text `ON` or `OFF` (always-on run / stop).
+
+### Subdevice Run Modes (Home Assistant)
+Controllable subdevices (WFC01/WFC02/AC1100) expose duration/volume run controls in addition to the on/off switch. The controller **stages** the selected mode and value, then fires the run when the Start button is pressed — the device self-closes at the duration/volume target.
+
+- **`<base>/+/subdevices/+/cmd/mode`** — selected run mode. Payload: `Duration` or `Volume`.
+- **`<base>/+/subdevices/+/cmd/set/duration`** — run duration in minutes (plain integer).
+- **`<base>/+/subdevices/+/cmd/set/volume`** — run volume in liters (plain integer; flow-capable valves only).
+- **`<base>/+/subdevices/+/cmd/start`** — start a run using the currently staged mode + value.
+
+All subdevice command topics (the four above plus `cmd` and `cmd/homeassistant`) are covered by a single `<base>/+/subdevices/+/cmd/#` subscription. A run mode the device doesn't support (e.g. `Volume` on a flow-less valve) is rejected; a `Start` with a missing/invalid value falls back to a safe default (3 min / 5 L).
 
 ## Home Assistant Discovery
 
@@ -80,6 +93,9 @@ When `controller.homeassistantdiscovery` is enabled, discovery configs are publi
 - **`homeassistant/sensor/<device_name>_<sensor_name>/config`** — Sensor entities
 - **`homeassistant/binary_sensor/<device_name>_<sensor_name>/config`** — Binary sensor entities (rain state, running state, etc.)
 - **`homeassistant/switch/<subdevice_nickname>/config`** — Switch entities for controllable subdevices
+- **`homeassistant/number/<subdevice_nickname>_<param>/config`** — Run Duration / Run Volume inputs (`entity_category: config`)
+- **`homeassistant/select/<subdevice_nickname>/config`** — Run Mode selector (published only when 2+ modes apply)
+- **`homeassistant/button/<subdevice_nickname>/config`** — Start Run button
 
 ### Discovery Payload Structure
 

src/EcoWitt.Controller.Tests/DiscoveryBuilderTest.cs

@@ -142,6 +142,51 @@ public void BuildSwitchConfig()
         Assert.That(config.StateTopic, Does.Contain("running"));
     }
 
+    [Test]
+    public void BuildNumberConfig_SetsRangeAndCommandTopic()
+    {
+        var device = DiscoveryBuilder.BuildDevice("valve");
+        var origin = DiscoveryBuilder.BuildOrigin();
+
+        var cfg = DiscoveryBuilder.BuildNumberConfig(device, origin, "Run Duration", "id_dur",
+            commandTopic: "ecowitt/gw/subdevices/1/cmd/set/duration",
+            min: 1, max: 1440, step: 1, unitOfMeasurement: "min", entityCategory: "config");
+
+        Assert.That(cfg.CommandTopic, Is.EqualTo("ecowitt/gw/subdevices/1/cmd/set/duration"));
+        Assert.That(cfg.Min, Is.EqualTo(1));
+        Assert.That(cfg.Max, Is.EqualTo(1440));
+        Assert.That(cfg.Step, Is.EqualTo(1));
+        Assert.That(cfg.UnitOfMeasurement, Is.EqualTo("min"));
+        Assert.That(cfg.SensorCategory, Is.EqualTo("config"));
+    }
+
+    [Test]
+    public void BuildSelectConfig_SetsOptions()
+    {
+        var device = DiscoveryBuilder.BuildDevice("valve");
+        var origin = DiscoveryBuilder.BuildOrigin();
+
+        var cfg = DiscoveryBuilder.BuildSelectConfig(device, origin, "Run Mode", "id_mode",
+            commandTopic: "ecowitt/gw/subdevices/1/cmd/mode",
+            options: new List<string> { "Duration", "Volume" });
+
+        Assert.That(cfg.Options, Is.EqualTo(new List<string> { "Duration", "Volume" }));
+        Assert.That(cfg.CommandTopic, Is.EqualTo("ecowitt/gw/subdevices/1/cmd/mode"));
+    }
+
+    [Test]
+    public void BuildButtonConfig_SetsPayloadPress()
+    {
+        var device = DiscoveryBuilder.BuildDevice("valve");
+        var origin = DiscoveryBuilder.BuildOrigin();
+
+        var cfg = DiscoveryBuilder.BuildButtonConfig(device, origin, "Start Run", "id_start",
+            commandTopic: "ecowitt/gw/subdevices/1/cmd/start", payloadPress: "PRESS");
+
+        Assert.That(cfg.PayloadPress, Is.EqualTo("PRESS"));
+        Assert.That(cfg.CommandTopic, Is.EqualTo("ecowitt/gw/subdevices/1/cmd/start"));
+    }
+
     // Device class mapping
     [TestCase(SensorType.Temperature, "temperature")]
     [TestCase(SensorType.Humidity, "humidity")]

src/EcoWitt.Controller.Tests/DispatcherGuardTest.cs

@@ -0,0 +1,62 @@
+using Ecowitt.Controller.Model.Api;
+using Ecowitt.Controller.Service.Orchestrator;
+using NUnit.Framework;
+
+namespace EcoWitt.Controller.Tests;
+
+[TestFixture]
+public class DispatcherGuardTest
+{
+    [Test]
+    public void Guard_DurationUnset_DefaultsTo3Min()
+    {
+        var cmd = new SubdeviceApiCommand { Cmd = Command.Start, Id = 1, Unit = DurationUnit.Minutes, Duration = null };
+        Dispatcher.GuardRunCommand(cmd);
+        Assert.That(cmd.Duration, Is.EqualTo(3));
+        Assert.That(cmd.AlwaysOn, Is.Not.EqualTo(true));
+    }
+
+    [Test]
+    public void Guard_NegativeDuration_DefaultsTo3Min()
+    {
+        var cmd = new SubdeviceApiCommand { Cmd = Command.Start, Id = 1, Unit = DurationUnit.Minutes, Duration = -1 };
+        Dispatcher.GuardRunCommand(cmd);
+        Assert.That(cmd.Duration, Is.EqualTo(3));
+    }
+
+    [Test]
+    public void Guard_HugeDuration_ClampedToMax()
+    {
+        var cmd = new SubdeviceApiCommand { Cmd = Command.Start, Id = 1, Unit = DurationUnit.Minutes, Duration = int.MaxValue };
+        Dispatcher.GuardRunCommand(cmd);
+        Assert.That(cmd.Duration, Is.EqualTo(1440));
+    }
+
+    [Test]
+    public void Guard_VolumeUnset_DefaultsTo5L()
+    {
+        var cmd = new SubdeviceApiCommand { Cmd = Command.Start, Id = 1, Unit = DurationUnit.Liters, Duration = 0 };
+        Dispatcher.GuardRunCommand(cmd);
+        Assert.That(cmd.Duration, Is.EqualTo(5));
+    }
+
+    [Test]
+    public void Guard_BareStartNoUnit_StaysAlwaysOn()
+    {
+        // The HA switch ON: Start, no Unit, no Duration → always-on, NOT defaulted to a duration run.
+        var cmd = new SubdeviceApiCommand { Cmd = Command.Start, Id = 1, Unit = null, Duration = null };
+        Dispatcher.GuardRunCommand(cmd);
+        Assert.That(cmd.AlwaysOn, Is.True);
+        Assert.That(cmd.Duration, Is.Null);
+    }
+
+    [Test]
+    public void Guard_Stop_Untouched()
+    {
+        var cmd = new SubdeviceApiCommand { Cmd = Command.Stop, Id = 1 };
+        Dispatcher.GuardRunCommand(cmd);
+        Assert.That(cmd.AlwaysOn, Is.Not.EqualTo(true));
+        Assert.That(cmd.Duration, Is.Null);
+    }
+
+}