diff --git a/.ebus-spec.json b/.ebus-spec.json index 9f3a4d9..993ac11 100644 --- a/.ebus-spec.json +++ b/.ebus-spec.json @@ -19,7 +19,6 @@ "mqtt-homie-transport", "tls", "mtls-client-auth", - "mdns-discovery", "proxy-publishers" ], "implements": { @@ -37,5 +36,5 @@ "devices": { "utility-meter": "0.6" }, "registries": { "capability-types": "0.19", "device-types": "0.5" } }, - "notes": "role=library: the SDK provides the framework mechanisms (supports). Excluded framework features: broker-hosting (broker's concern, not the SDK) and rest-configuration (not implemented). mdns-discovery is provided via the optional `mdns` extra and the examples. The implements map covers the examples/utility-meter reference publisher plus the capability vocabulary the SDK itself encodes. grid/status/demand/power-quality were previously defined inline in the utility-meter data model and covered implicitly by pinning it; utility-meter 0.6 split them out into standalone catalogs, so they are now pinned directly. soc is pinned because the ebus_sdk.ha customizer maps its properties to Home Assistant metadata and examples/ha-discovery-bridge publishes a soc node; it is a capability the SDK has first-class knowledge of rather than merely transports. If the simulators migrate out of examples/ into their own repo(s), each would carry its own application lockfile and this implements map would reduce to the SDK proper." + "notes": "role=library: the SDK provides the framework mechanisms (supports). Excluded framework features: broker-hosting (broker's concern, not the SDK), rest-configuration (not implemented) and mdns-discovery (not implemented in the library; the optional `mdns` extra only installs zeroconf for the examples that browse for a broker). The implements map covers the examples/utility-meter reference publisher plus the capability vocabulary the SDK itself encodes. grid/status/demand/power-quality were previously defined inline in the utility-meter data model and covered implicitly by pinning it; utility-meter 0.6 split them out into standalone catalogs, so they are now pinned directly. soc is pinned because the ebus_sdk.ha customizer maps its properties to Home Assistant metadata and examples/ha-discovery-bridge publishes a soc node; it is a capability the SDK has first-class knowledge of rather than merely transports. If the simulators migrate out of examples/ into their own repo(s), each would carry its own application lockfile and this implements map would reduce to the SDK proper." } diff --git a/CHANGELOG.md b/CHANGELOG.md index 0357a64..f7a04e7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,41 @@ All notable changes to `ebus-sdk` are recorded here. Format follows [Keep a Chan ## [Unreleased] +Homie 5 compliance, phase 1: producer wire correctness ([#95](https://github.com/electrification-bus/python-sdk/issues/95), part of [#91](https://github.com/electrification-bus/python-sdk/issues/91)). Line numbers refer to [homieiot/convention@7edc221](https://github.com/homieiot/convention/blob/7edc221336f1644a9f04445cbc50c5af695bb047/convention.md). + +### Changed + +- `Device.stop()` publishes `$state=disconnected` for every device in the tree, descendants first and the root last, before its clean disconnect. Only the root announced it before, so each child's retained `$state` stayed `ready` after a clean shutdown (convention.md:280-281; the cascade at :262-267 covers only `lost`). +- `Device.stop(announce=False)` no longer disconnects cleanly. On an SDK-owned client it ends the connection without an MQTT DISCONNECT, so the broker publishes the Last Will. A clean disconnect must be preceded by `disconnected` (:281), and `lost` is the state of a bad disconnect (:284). After `declare_lost()` the will re-asserts the same retained `lost`; unpaired, it replaces the stale `ready` the clean disconnect used to leave. +- `child.delete()` updates the parent first (`init`, `$description` without the child, `ready`) and only then clears the child's topics, starting with `$state` (:635-641). The order was reversed. +- `Device.will()` carries `qos` (the tree's) and `retain: True`, so the Last Will is no longer sent at QoS 0 while every other `$state` uses the tree QoS (:45, :689-691). +- `$description` `version` is the top 52 bits of the SHA-256 of the document without `version` (the content hash the publish gate already used), replacing epoch milliseconds. Every change gets a new version, two documents published in one millisecond no longer share one, and an unchanged document keeps its version, so a reconnect republishes a byte-identical document (:215). +- A `$description` change on a `sleeping` device is wrapped in `init` and returns to `sleeping`, as a change on a `ready` device already was (:207). A reconnect republish of unchanged content is not a change and publishes no `init`. When the content differs from the last document published (after `invalidate_description_cache()`, say), `refresh_tree()` publishes `init` first and the device's state once, after its values and children. +- `Node.add_property()`, `Device.add_node()`, `Node.delete_property()`, `Device.remove_node()` and `Device.delete_node()` called outside a `state_transition()` on a `ready` or `sleeping` device publish `init`, the changed values, the new `$description`, then the prior state. `ready` was announced before a new property's value existed (:279, :611). A change that leaves the `$description` as last published (a property added to a node not yet attached to its device, or an identical re-declaration) publishes only its values, with no `init`/`ready` flap (:616). +- `$description` omits `type` for a device or node that has none instead of publishing `"type": null` (:218, :305), and a property's `retained` is always a boolean. +- A `json` property whose `format` is a dict publishes it as a JSON string, not a nested object (:395). +- Invalid producer input is warned about, once per object, and still published; it will raise in the next minor release: a property with no `datatype` (published without one, :340), `retained=None` (published as `false`, the non-retained behavior it always had, :343), an `enum` or `color` with no `format`, an enum format with an empty or duplicate value (:392-393), and a `color` format listing anything but `rgb`, `hsv`, `xyz`. +- `description_extras` keys that are Homie core `$description` fields (`homie`, `version`, `nodes`, `name`, `type`, `children`, `root`, `parent`, `extensions`) are dropped with a warning. An extra could add `root` and `parent` to a root device (:220). +- An outbound value the property's datatype cannot represent is refused: nothing is published, `set_value()` returns `False`, and a `reason=propertyValueRefused` warning is logged. Refused: non-finite floats (:99), a `bool` on `integer` or `float`, an integer that is not whole or outside 64 bits (:87-89), an enum value not in `format` (:111), a `float` beyond the 64-bit float range (:95-96), a `datetime` string that is not ISO 8601 (:132), and a `json` value that is not an array or object or contains NaN (:150). Floats are written without `+` in the exponent (:97). Payload digits are ASCII `0-9` only (:89, :97), so a string carrying another script's digits is refused on `integer`, `float`, `duration`, `datetime` and `color`. A `decimal.Decimal` is accepted on `float`, and a non-`float` real (numpy `float32`, say) keeps its own text when that is already a valid payload, so a rounded `float32` publishes `0.1` and not `0.10000000149011612`. +- The `0x00` empty-string encoding applies only to `string` properties (:65-67); `""` on another datatype is refused. +- A non-retained (event) property publishes non-retained at QoS 0 (:50). The reconnect and structural-change republish walks skip it, so the last event is no longer replayed (:695), and clearing it publishes nothing instead of a retained empty message on the event topic. +- `Device.publish("$alert", ...)` publishes nothing and warns: an alert needs an id (:517). Use `publish_alert()`. +- An inbound `/set` is validated for the property's datatype before the callback runs, and an invalid one is dropped with a `reason=propertySetRejected` warning: `json` must be an array or object, `integer` and `float` must follow the payload grammar, a number is rounded to the format's step and then checked against min/max (:399), an `enum` value must be in `format`, a `datetime` must be ISO 8601, and a `boolean` must be `true` or `false`. A number that rounding changed reaches the callback in its rounded form. `0x00` decodes to `""` only for `string`. +- A `/set` delivered with the retain flag is ignored with a warning: it is a stale command replayed at subscribe time (:48, :483). This uses `subscribe(..., with_retain=True)` from ebus-mqtt-client 0.7.0 ([electrification-bus/ebus-mqtt-client#24](https://github.com/electrification-bus/ebus-mqtt-client/issues/24)); the dependency floor rises to `ebus-mqtt-client>=0.7.0`. An injected transport without that parameter skips the check, and a debug line says so. +- `supports_target=True` is inert: one warning at construction, and `/set` no longer calls the `$target` stub, which ran before the payload was accepted (:466, :470). +- `.ebus-spec.json` no longer lists `mdns-discovery` under `supports`; the library has no mDNS code. +- `examples/simple-device`, `simple-tree-device` and `utility-meter` call `stop()` on Ctrl-C. They exited without it, so the broker published the Last Will and the tree was left with its root `lost` and its children `ready`. + +### Added + +- Encoders for `datetime` (`datetime`/`date` as ISO 8601, :132), `duration` (`timedelta` as `PTxHxMxS`, :137-144) and `color` (a component tuple in the first `format` color type, :118-123). These fell through to `str()`, which produced `2026-10-08 12:00:00+00:00`, `0:05:00` and `(255, 0, 0)`. +- `Device.publish_alert(alert_id, message)`, `Device.clear_alert(alert_id)` and `Device.alerts()`. Alerts publish retained to `$alert/[alert ID]` (:517), are republished on reconnect, and are cleared by `delete()` (:289). +- `Device.invalidate_description_cache()`, the `$description` counterpart of `Property.invalidate_publish_cache()`. `clear_retained_topic()` aimed at the device's own `$description` and `delete_all_from_mqtt()` call it, so a later `publish_description()` is no longer suppressed after the topic was cleared. + +### Fixed + +- `delete()` followed by `stop()` re-created the deleted device as a bare `$state=disconnected` with no `$description` (:272, :288). After `delete()` the root's `stop()` announces nothing, flushes `delete()`'s retained clears and disconnects cleanly in either mode, and a reconnect's `refresh_tree()` republishes nothing. + ## [0.24.0] — 2026-10-03 ### Fixed diff --git a/README.md b/README.md index 7990aee..542cb28 100644 --- a/README.md +++ b/README.md @@ -67,12 +67,12 @@ Inject the client before it connects, then wire the two Homie-correctness pieces client = my_host_mqtt_client() # created, not yet connected device = Device('panel-1', type='...', mqttc=client) -client.set_will(**device.will()) # LWT ($state=lost); must precede connect +client.set_will(**device.will()) # LWT ($state=lost, tree QoS, retained); must precede connect client.on_connect(device.refresh_tree) # re-announce the retained tree on every (re)connect client.connect() # host connects on its own loop ``` -`device.will()` returns the tree's Last Will descriptor and `device.refresh_tree()` republishes the whole tree; the `set_will` / `on_connect` / `connect` calls above are illustrative of your host's own MQTT API. Property values publish once the client is connected (the SDK gates on `is_connected()`, not on its own `start()`, which a caller-driven client never calls). `device.stop()` publishes a final retained `$state=disconnected` through the client and returns immediately, without flushing or closing it; `device.stop(announce=False)` publishes nothing and leaves the retained `$state` as it stands, for a caller that published its own final state first (see `declare_lost()` below). `on_disconnect=` is inert for an injected client; register disconnect handling on your own client. +`device.will()` returns the tree's Last Will descriptor and `device.refresh_tree()` republishes the whole tree; the `set_will` / `on_connect` / `connect` calls above are illustrative of your host's own MQTT API. Property values publish once the client is connected (the SDK gates on `is_connected()`, not on its own `start()`, which a caller-driven client never calls). `device.stop()` publishes a final retained `$state=disconnected` for every device in the tree through the client and returns immediately, without flushing or closing it; `device.stop(announce=False)` publishes nothing, for a caller that published its own final state first (see `declare_lost()` below), and that caller then drops the connection without a clean DISCONNECT so the will fires. `on_disconnect=` is inert for an injected client; register disconnect handling on your own client. For the inbound direction, if the tree has settable properties whose callbacks are async coroutines, pass `Device(async_loop=)`: inbound `/set` arrives on the transport's network thread, and this schedules the callback onto your loop (set once for the whole tree, not per property). A synchronous callback runs inline and needs no loop. @@ -105,7 +105,7 @@ A device has three ways to stop, and they mean different things to a consumer re | Teardown | `$state` left retained | How | | --- | --- | --- | -| Graceful shutdown | `disconnected` | `device.stop()` | +| Graceful shutdown | `disconnected` on every device in the tree | `device.stop()` | | Ungraceful death (crash, power loss) | `lost` | the Last Will, which fires only on an *unclean* disconnect | | Deliberate death | `lost` | `device.declare_lost()` | @@ -113,27 +113,49 @@ The third is for a producer that knows it is failing: a fatal error handler, a s ```python device.declare_lost() # root's $state=lost, published and (owned path) flushed -device.stop(announce=False) # tear down without overwriting it with `disconnected` +device.stop(announce=False) # drop the connection without a clean DISCONNECT; the will re-asserts `lost` ``` `declare_lost()` is **tree-level**, like `will()` and `stop()`: it publishes the *root's* `$state`, which per the Homie 5 effective-state rule makes every descendant lost too, and it publishes exactly the topic and payload `will()` describes so the two paths cannot drift. To mark one device lost (a proxy whose single upstream vanished), use `set_state(DeviceState.LOST)` on that device instead. It moves the state and publishes it together, and the move is unconditional: publishing a state the `Device` does not hold is how a later `refresh_tree()` silently republishes `ready` over it. It returns whether `$state` actually moved, the same convention as `set_state`: on an injected transport, a `True` from a connected tree is your cue to drain, and `False` means the root was already lost. It is not a delivery signal and cannot be one there. Publishing is skipped entirely when the broker is unreachable (the state still moves, and the next connect republishes it), so `True` does not by itself prove anything was queued. It does not stop the client. -`stop(announce=False)` unpaired leaves whatever was published last, typically a stale `ready`, and nothing will correct it: the clean disconnect `stop()` performs suppresses the LWT. Neither call substitutes for the will, because a crashed process calls nothing. +`stop()` publishes `disconnected` for every device in the tree, descendants first and the root last, before its clean disconnect: Homie 5 requires each device to send `disconnected` before cleanly disconnecting, and the root-to-child cascade covers only `lost`. `stop(announce=False)` publishes nothing, so it does not disconnect cleanly either: on an SDK-owned client it ends the connection without an MQTT DISCONNECT and the broker publishes the Last Will (`lost`). Paired with `declare_lost()` that re-asserts the same retained payload; unpaired, the will replaces the stale `ready` the clean disconnect used to leave behind. On an injected client the SDK closes nothing, so drop your connection without a clean DISCONNECT to get the same result. After `delete()` on the root, `stop()` announces nothing in either mode and disconnects cleanly, since any later `$state` would re-create the deleted device. Neither call substitutes for the will, because a crashed process calls nothing. #### Clearing a value vs. an empty-string value Homie 5 distinguishes two things that both look "empty" on the wire, and the SDK handles each automatically: - **Clearing (retracting) a retained value** — set the property to `None`. Once it has been published, this emits a zero-length `retain=True` payload, which MQTT/Homie treats as "delete the retained topic", so a subscriber that connects later sees no stale value. (`clear_value()` does the same explicitly; `Node.delete_property()` clears on removal.) A `None` that was never published is a silent no-op — no phantom topic is created. -- **An actual empty-string value** — set a string property to `""`. This is published as a single null byte (`0x00`), the Homie 5 encoding that keeps `""` distinct from a topic-clear. Inbound `0x00` payloads are decoded back to `""` on the controller and on `/set`. Helpers `encode_empty_string()` / `decode_empty_string()` and the constant `HOMIE_EMPTY_STRING_PAYLOAD` are exported for consumers that need them directly. +- **An actual empty-string value** — set a `string` property to `""`. This is published as a single null byte (`0x00`), the Homie 5 encoding that keeps `""` distinct from a topic-clear. Only `string` has an empty value: `""` on any other datatype is refused like any other invalid value (see below). Inbound `0x00` payloads are decoded back to `""` on the controller and, for a `string` property, on `/set`. Helpers `encode_empty_string()` / `decode_empty_string()` and the constant `HOMIE_EMPTY_STRING_PAYLOAD` are exported for consumers that need them directly. ```python temp.set_value(None) # retracts the retained topic (subscribers see nothing) label.set_value("") # publishes an empty-string VALUE (0x00 on the wire) ``` +#### Values the datatype cannot represent are refused + +A value is encoded for its property's datatype before it is published (convention.md:74-150 at [homieiot/convention@7edc221](https://github.com/homieiot/convention/blob/7edc221336f1644a9f04445cbc50c5af695bb047/convention.md)). A value the datatype cannot represent is refused: nothing is published, the retained topic keeps the last good value, `set_value()` returns `False`, and a `reason=propertyValueRefused` warning names the problem. Refused: NaN and infinities on `float`, a `bool` on `integer` or `float`, an `integer` that is not whole or does not fit in 64 bits, an `enum` value not in its `format`, a `float` beyond the 64-bit float range, a `json` value that is not an array or object or contains NaN, and `""` on anything but `string`. Floats are written without `+` in the exponent (`1e20`, not `1e+20`). + +Native values are encoded for you: `datetime`/`date` as ISO 8601 (`2026-10-08T12:00:00+00:00`), `timedelta` as `PTxHxMxS` (`PT12H5M46S`; days fold into hours), and a color tuple as the property's first `format` color type plus its components (`(255, 0, 0)` with `format="rgb,hsv"` publishes `rgb,255,0,0`). A string is checked against the datatype's payload grammar and published as given; for `datetime` that is ISO 8601 (calendar, ordinal or week date, extended or basic form, with an optional time and zone), so `2026-10-08 12:00` is refused. Payload digits are ASCII `0-9`. + +#### Event (non-retained) properties + +A property with `retained=False` publishes non-retained at QoS 0, as Homie 5 requires for events: a subscriber gets the event now or not at all. A reconnect's `refresh_tree()` and a structural change (`add_node`, `add_property`) never replay an event's last value, and setting it to `None` publishes nothing, since the broker holds nothing to retract. + +#### Alerts + +`device.publish_alert("battery", "Battery is low, at 8%")` raises a retained alert at `$alert/battery`; `device.clear_alert("battery")` deletes it. Alerts are republished on reconnect and cleared by `delete()`. `publish("$alert", ...)` without an alert id publishes nothing and logs a warning. + +#### Inbound `/set` is validated + +A `/set` payload is checked against the property's datatype before its `set_callback` runs, and an invalid one is dropped with a `reason=propertySetRejected` warning: `json` must be an array or object (and match a `$format` JSONschema when one can be checked), `integer` and `float` must follow the payload grammar, `boolean` must be `true` or `false`, an `enum` value must be in its `format`, and a `datetime` must be ISO 8601. A number is rounded to its format's step and then checked against min/max (convention.md:397-413): the callback receives the payload as sent, or its rounded form if rounding changed it (`5` on `0:10:2` arrives as `6`). + +A `/set` the broker delivers with the retain flag set is a stale command replayed at subscribe time, since controllers publish `/set` non-retained only; it is ignored with a `reason=propertySetRetainedIgnored` warning. That needs ebus-mqtt-client 0.7.0 or later, whose `subscribe(..., with_retain=True)` delivers the flag. With an older client, or an injected transport whose `subscribe` has no `with_retain` parameter, the check is skipped and a debug line says so. + +`supports_target=True` is inert: `$target` is not implemented, so the property logs one warning at construction and never publishes `$target`. + #### Unchanged values are not republished Setting a retained property to a value whose wire payload is byte-identical to the one it last published is a no-op: the broker's retained store already holds that payload, every subscriber already has it, and the publish carries no information. The comparison is on the **final** payload, after rounding, datatype coercion and empty-string encoding, which is why the SDK does this rather than leaving it to the caller: a caller holding a raw reading cannot know whether it will change the payload, since `round_to` and the coercion live inside the property. Two readings of `0.14494210481643677` and `0.14501120000000001` are genuinely different values that any caller-side change check calls "changed", and both publish `0.1`. @@ -142,7 +164,7 @@ Three carve-outs, each deliberate: - **A non-retained (event) property is never gated.** The broker stores nothing for it, so an identical consecutive payload is a second real event, not a redundant write. - **Retraction always publishes.** `set_value(None)` / `clear_value()` must reach the broker to delete the topic. -- **Every whole-tree republish forces.** `refresh_tree()`, and the reconnect that calls it, republish every property value regardless. That is what repopulates a broker whose retained store is empty (restarted without persistence, or a fresh one); a gated refresh would find every payload equal to what it last published and send nothing. +- **Every whole-tree republish forces.** `refresh_tree()`, and the reconnect that calls it, republish every retained property value regardless. That is what repopulates a broker whose retained store is empty (restarted without persistence, or a fresh one); a gated refresh would find every payload equal to what it last published and send nothing. The retained state left on the broker is identical either way: strictly fewer messages, same truth. `Property.get_last_published_value()` returns the memoized wire payload, and `Property.invalidate_publish_cache()` forgets it, which is what you call if something deletes a retained value topic behind the SDK's back. @@ -172,7 +194,7 @@ panel.children()[0].delete() Children may have children of their own. A single Last Will registered on the root marks the entire tree `lost` if the publisher process dies, and `root.declare_lost()` publishes the same thing deliberately when the publisher knows it is dying — controllers compute effective state per the Homie 5 precedence table (see [`HOMIE_EFFECTIVE_STATE_TABLE`](src/ebus_sdk/homie.py)). -`$description` republishes are minimized: structural changes made inside one `state_transition()` collapse to a single consolidated publish at exit (not one per `add_node`), and `publish_description()` is a no-op when the description content (ignoring its `version` timestamp) is unchanged — so a `state_transition()` that changes nothing structural does not re-emit the (potentially multi-KB) `$description`. A reconnect always republishes regardless, to restore retained state. Note this suppresses the redundant `$description` payload, not the `$state` `init`→`ready` edge of an empty transition. Property *values* are minimized the same way and with the same reconnect carve-out (see [Unchanged values are not republished](#unchanged-values-are-not-republished)). +`$description` republishes are minimized: structural changes made inside one `state_transition()` collapse to a single consolidated publish at exit (not one per `add_node`), and `publish_description()` is a no-op when the description content is unchanged — so a `state_transition()` that changes nothing structural does not re-emit the (potentially multi-KB) `$description`. A reconnect always republishes regardless, to restore retained state. Note this suppresses the redundant `$description` payload, not the `$state` `init`→`ready` edge of an empty transition. Property *values* are minimized the same way and with the same reconnect carve-out (see [Unchanged values are not republished](#unchanged-values-are-not-republished)). The document's `version` is derived from its content (the top 52 bits of a SHA-256 of the document without `version`), so an unchanged document keeps its version across reconnects and restarts, and any change gets a new one. Homie 5 lets `$description` change only while the device is `init`, `disconnected` or `lost`: a structural change made outside a `state_transition()` on a `ready` or `sleeping` device is published as `init`, the changed values, the new `$description`, then the state the device was in. A change that leaves the document as last published (a property added to a node not yet attached to its device, or an identical re-declaration) publishes only its values. ### Publishing under a different Homie domain @@ -360,7 +382,7 @@ See [`examples/README.md`](examples/README.md) for example scripts demonstrating ## Requirements - Python 3.10+ -- [`ebus-mqtt-client`](https://github.com/electrification-bus/ebus-mqtt-client) >= 0.6.0 (the MQTT transport layer; it pins `paho-mqtt`, so the SDK does not depend on paho directly. 0.6.0 holds every publish issued while the link is down and flushes it before `on_connect`, where 0.5.0 left QoS 1 and 2 to paho, whose replay landed after the SDK's on-connect republish and could exceed the broker's receive quota, leaving a root's retained `$state` on `init`; 0.5.0 holds retained QoS 0 publishes issued before the link is up and flushes them on connect, rather than dropping them and warning about each: a root built with `mqtt_cfg=` connects asynchronously, so the SDK itself creates that window whenever a construction-time `state_transition()` publishes before CONNACK; 0.4.0 provides `MqttClient.asyncio_driver()`, the loop-native alternative to paho's background thread; 0.3.0 ships the `py.typed` marker, so a downstream type checker resolves the re-exported `MqttClient` to the concrete class rather than `Any`; 0.2.0 adds the `on_disconnect_callback` the SDK's disconnect hook adopts; and, since 0.1.8, it carries the asynchronous, down-broker-tolerant connect the resilient-connect behavior relies on) +- [`ebus-mqtt-client`](https://github.com/electrification-bus/ebus-mqtt-client) >= 0.7.0 (the MQTT transport layer; it pins `paho-mqtt`, so the SDK does not depend on paho directly. 0.7.0 exposes the MQTT retain flag, so a device ignores a retained `/set` replayed at resubscribe; 0.6.0 holds every publish issued while the link is down and flushes it before `on_connect`, where 0.5.0 left QoS 1 and 2 to paho, whose replay landed after the SDK's on-connect republish and could exceed the broker's receive quota, leaving a root's retained `$state` on `init`; 0.5.0 holds retained QoS 0 publishes issued before the link is up and flushes them on connect, rather than dropping them and warning about each: a root built with `mqtt_cfg=` connects asynchronously, so the SDK itself creates that window whenever a construction-time `state_transition()` publishes before CONNACK; 0.4.0 provides `MqttClient.asyncio_driver()`, the loop-native alternative to paho's background thread; 0.3.0 ships the `py.typed` marker, so a downstream type checker resolves the re-exported `MqttClient` to the concrete class rather than `Any`; 0.2.0 adds the `on_disconnect_callback` the SDK's disconnect hook adopts; and, since 0.1.8, it carries the asynchronous, down-broker-tolerant connect the resilient-connect behavior relies on) Optional extras: diff --git a/doc/building-a-proxy.md b/doc/building-a-proxy.md index c34ed63..657b78e 100644 --- a/doc/building-a-proxy.md +++ b/doc/building-a-proxy.md @@ -206,7 +206,7 @@ Each `add()` announces its own device and makes the parent republish its `$descr - **Batch structural changes.** Adding N nodes/properties inside one `with device.state_transition():` collapses to a single `$description` publish and one `init` to `ready` edge, instead of N. Always build a device's structure inside a transition. - **Connect before you publish.** `Device(..., mqtt_cfg=...)` connects asynchronously. If you build and publish before the broker connection is established, the first retained `$description` / `$state` the broker keeps can be a pre-connect snapshot until the SDK's on-connect refresh corrects it. Wait for `device.mqttc.is_connected()` before the initial build so the first retained state is correct. -- **Drive `$state` from availability.** When your upstream reports ONE device offline, `set_state(DeviceState.LOST)` on that child (and `READY` when it returns). When the whole bridge is dying, `root.declare_lost()` publishes the root's `$state=lost`, which per the Homie 5 effective-state rule covers every descendant in a single publish; follow it with `stop(announce=False)` so the teardown does not overwrite it with `disconnected`. Do not reach for `declare_lost()` for one dead upstream: it blanks the entire tree's liveness. The root's Last Will still covers process death, which neither call can, since a crashed process calls nothing. +- **Drive `$state` from availability.** When your upstream reports ONE device offline, `set_state(DeviceState.LOST)` on that child (and `READY` when it returns). When the whole bridge is dying, `root.declare_lost()` publishes the root's `$state=lost`, which per the Homie 5 effective-state rule covers every descendant in a single publish; follow it with `stop(announce=False)`, which does not overwrite it with `disconnected` and ends the connection without a clean DISCONNECT, so the broker's will re-asserts the same `lost`. Do not reach for `declare_lost()` for one dead upstream: it blanks the entire tree's liveness. The root's Last Will still covers process death, which neither call can, since a crashed process calls nothing. ## Settable / bidirectional properties (control back to the device) @@ -224,7 +224,7 @@ If you build the tree by hand instead of via `build_from_declarations`, wire the ### Settable `json` properties and `$format` validation -For a settable `json` property (a compound command like `flex/request`), give the `PropertySpec` a `format` that is the JSON Schema of the command surface your device accepts. An inbound `/set` payload is then `json.loads`ed to a `dict`/`list` and validated against that schema before your `entity_setter` runs, so your `entity_setter` receives a parsed, schema-valid object and a malformed or out-of-surface command is rejected for you: +For a settable `json` property (a compound command like `flex/request`), give the `PropertySpec` a `format` that is the JSON Schema of the command surface your device accepts. An inbound `/set` payload is then `json.loads`ed, required to be a `dict`/`list` (Homie 5 allows only an array or object), and validated against that schema before your `entity_setter` runs, so your `entity_setter` receives a parsed, schema-valid object and a malformed or out-of-surface command is rejected for you: ```python PropertySpec( diff --git a/examples/simple-device b/examples/simple-device index 1aac4cc..1d69cd2 100755 --- a/examples/simple-device +++ b/examples/simple-device @@ -19,7 +19,6 @@ from pprint import pformat class ExampleDevice: - def __init__(self, logger: Optional[logging.Logger] = None): # Set up logger if logger is None: @@ -29,9 +28,7 @@ class ExampleDevice: # Prevent duplicate handlers if multiple instances are created if not logger.handlers: handler = logging.StreamHandler() - formatter = logging.Formatter( - "%(asctime)s - %(name)s - %(levelname)s - %(message)s" - ) + formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") handler.setFormatter(formatter) logger.addHandler(handler) self._logger = logger @@ -54,9 +51,7 @@ class ExampleDevice: def get_property(self, group: str, property_id: str) -> Property: return self._props.get(group, property_id) - def add_property_on_change_callback( - self, group: str, id: str, callback: Callable - ) -> uuid.UUID: + def add_property_on_change_callback(self, group: str, id: str, callback: Callable) -> uuid.UUID: """ Adds a callback to the Property group.id, that will be called when the Property's value changes Returns a callback_id (a uuid1) that can be used subsequently to remove the callback @@ -73,7 +68,6 @@ class ExampleDevice: class ExampleDeviceAdapter: - def __init__( self, example_device: Any, @@ -88,9 +82,7 @@ class ExampleDeviceAdapter: # Prevent duplicate handlers if multiple instances are created if not logger.handlers: handler = logging.StreamHandler() - formatter = logging.Formatter( - "%(asctime)s - %(name)s - %(levelname)s - %(message)s" - ) + formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") handler.setFormatter(formatter) logger.addHandler(handler) self._logger = logger @@ -171,12 +163,8 @@ class ExampleDeviceAdapter: for homie_property_id, mapping_dict in properties.items(): group = mapping_dict.get("group", "sensors") py_property_id = mapping_dict.get("py_property_id", None) - self._logger.debug( - f"reason=addHomieEnvironment,pyPropertyId={py_property_id}" - ) - self._logger.info( - f"reason=addHomieEnvironmentNode,group={group},pyPropertyId={py_property_id}" - ) + self._logger.debug(f"reason=addHomieEnvironment,pyPropertyId={py_property_id}") + self._logger.info(f"reason=addHomieEnvironmentNode,group={group},pyPropertyId={py_property_id}") property = self._example_device.get_property(group, py_property_id) property_type = property.type() homie_datatype_from_property = None @@ -205,15 +193,17 @@ class ExampleDeviceAdapter: ) return homie_node + def stop(self) -> None: + """Announce `disconnected` for the device and disconnect cleanly.""" + self._homie_device.stop() + def create_device_and_nodes(self): self._homie_device = self.create_homie_device() self.add_homie_environment_node() def main(): - parser = argparse.ArgumentParser( - description="Example Homie device that publishes sensor data" - ) + parser = argparse.ArgumentParser(description="Example Homie device that publishes sensor data") parser.add_argument( "-c", "--config", @@ -225,17 +215,13 @@ def main(): config_file = args.config or os.environ.get("EBUS_BROKER_CFG") if not config_file: print("Error: No broker config file specified.") - print( - "Provide config via --config option or EBUS_BROKER_CFG environment variable." - ) + print("Provide config via --config option or EBUS_BROKER_CFG environment variable.") print() print("Example config file format:") print(" {") print(' "host": "mqtt.example.com",') print(' "port": 1883,') - print( - ' "authentication": {"type": "USER_PASS", "username": "user", "password": "pass"}' - ) + print(' "authentication": {"type": "USER_PASS", "username": "user", "password": "pass"}') print(" }") sys.exit(1) @@ -254,13 +240,16 @@ def main(): example_device = ExampleDevice(logger=example_device_logger) example_device_logger.info(f"\n{pformat(example_device.as_dict())}") - example_device_adapter = ExampleDeviceAdapter( - example_device, mqtt_cfg, logger=example_device_adapter_logger - ) + example_device_adapter = ExampleDeviceAdapter(example_device, mqtt_cfg, logger=example_device_adapter_logger) example_device_adapter_logger.info(f"\n{pformat(example_device_adapter.as_dict())}") - # Wait forever for event that will never come - event = threading.Event() - event.wait() + # Wait until Ctrl-C, then shut down cleanly: stop() publishes + # $state=disconnected before disconnecting, where an exit without it leaves + # the broker to publish the Last Will (`lost`). + try: + threading.Event().wait() + except KeyboardInterrupt: + example_device_adapter.stop() + raise if __name__ == "__main__": diff --git a/examples/simple-tree-device b/examples/simple-tree-device index 4547388..ec4aa5a 100755 --- a/examples/simple-tree-device +++ b/examples/simple-tree-device @@ -138,14 +138,20 @@ def main() -> None: logger = logging.getLogger("simple-tree-device") mqtt_cfg = load_broker_config(config_file) - build_tree(mqtt_cfg, logger) + panel = build_tree(mqtt_cfg, logger) if args.check: logger.info("check mode: tree built, exiting") return logger.info("tree is live; Ctrl-C to exit. Watch it with: mosquitto_sub -t 'ebus/5/#' -v") - threading.Event().wait() + try: + threading.Event().wait() + except KeyboardInterrupt: + # stop() is tree-level: every device announces `disconnected`, children + # first and the root last, before the clean disconnect. + panel.stop() + raise if __name__ == "__main__": diff --git a/examples/utility-meter b/examples/utility-meter index 4e7b894..ab6efde 100755 --- a/examples/utility-meter +++ b/examples/utility-meter @@ -457,6 +457,11 @@ class UtilityMeterAdapter: self._homie_device: Optional[homie.Device] = None self._create_device_and_nodes() + def stop(self) -> None: + """Announce `disconnected` for the meter and disconnect cleanly.""" + if self._homie_device is not None: + self._homie_device.stop() + @staticmethod def load_broker_config(config_file: str) -> Dict[str, Any]: with open(config_file, "r") as fp: @@ -832,10 +837,9 @@ def main(): sys.exit(2) # Wire it up. Constructing the adapter creates the Homie device, its - # capability nodes, and starts the MQTT client (side effects we rely on); - # nothing else references it, so we don't bind the instance. + # capability nodes, and starts the MQTT client. meter = UtilityMeter(meter_cfg, logger=_make_logger("UtilityMeter")) - UtilityMeterAdapter( + adapter = UtilityMeterAdapter( meter, device_id=device_id, device_name=device_name, @@ -851,8 +855,13 @@ def main(): DEVICE_TYPE, ) - # Wait forever for an event that will never come - threading.Event().wait() + # Wait until Ctrl-C, then shut down cleanly: stop() publishes + # $state=disconnected before disconnecting. + try: + threading.Event().wait() + except KeyboardInterrupt: + adapter.stop() + raise if __name__ == "__main__": diff --git a/pyproject.toml b/pyproject.toml index 460ad13..74d14c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,7 +30,10 @@ classifiers = [ dependencies = [ # MQTT transport lives in ebus-mqtt-client, which pins paho-mqtt; the SDK # does not import paho directly, so paho is a transitive dependency (not - # declared here). Floor 0.6.0: it holds every publish issued while the link + # declared here). Floor 0.7.0: subscribe(..., with_retain=True) exposes the + # MQTT retain flag, which lets a device ignore a retained /set the broker + # replays at every resubscribe (Homie 5 :48/:483; a caller-supplied transport + # without it skips that check). Earlier floors, still relied on: 0.6.0 holds every publish issued while the link # is down (retained at any QoS, newest per topic) and flushes it before # on_connect, where 0.5.0 left QoS 1 and 2 to paho. paho replayed those on # CONNACK uncapped and after the SDK's on-connect refresh_tree, so an older @@ -58,7 +61,7 @@ dependencies = [ # on_disconnect_callback the disconnect hook adopts (SDK-al5); 0.1.8 added the # connect_async construction the resilient-connect behavior relies on (older # clients raised ConnectionRefusedError on a down broker). - "ebus-mqtt-client>=0.6.0", + "ebus-mqtt-client>=0.7.0", ] [project.optional-dependencies] diff --git a/src/ebus_sdk/homie.py b/src/ebus_sdk/homie.py index 99aef40..b8f54a1 100644 --- a/src/ebus_sdk/homie.py +++ b/src/ebus_sdk/homie.py @@ -28,7 +28,7 @@ homie / 5 / [device ID] / [node ID] / [property ID]; reported property values (for string types) homie / 5 / [device ID] / [node ID] / [property ID] / set; the topic to set properties (of string types) homie / 5 / [device ID] / [node ID] / [property ID] / $target; the target property value (for string types) - The SDK encodes "" as 0x00 on publish (Property.publish_value, Controller.set_property) and decodes + The SDK encodes "" as 0x00 on publish of a string value (Property.publish_value, Controller.set_property) and decodes 0x00 back to "" on receive (Controller._on_property_message / _on_target_message, Property._settable_callback). This convention specifies no way to represent an actual value of a 1-character string with a single byte 0. If a device needs this, then it should provide an escape mechanism on the application level. @@ -38,11 +38,20 @@ """ import asyncio +import contextlib +import datetime +import decimal import hashlib +import inspect import json import logging +import math +import numbers import os import re +import socket +import sys +import threading import time import uuid from enum import Enum @@ -56,6 +65,7 @@ class StrEnum(str, Enum): from dataclasses import dataclass +from fractions import Fraction from functools import partial from threading import RLock @@ -90,6 +100,20 @@ def _log_missing_client(message: str, *, by_design: bool, level: int = logging.W logger.log(logging.DEBUG if by_design else level, message) +def _warn_invalid_once(owner: Any, problem: str, message: str) -> None: + """Warn about invalid producer input once per object and problem. + + The policy for invalid producer input (#91): warn in this release, still + publish, and raise in the next minor release. Once per object keeps a tree + rebuilt on every reconnect from repeating itself. + """ + warned = owner.__dict__.setdefault("_invalid_input_warned", set()) + if problem in warned: + return + warned.add(problem) + logger.warning(f"{message},hint=invalid Homie 5 input; this will raise in the next minor release") + + # One-time warning when a `$format` JSONSchema is present but jsonschema is not. _jsonschema_warned = False @@ -167,6 +191,15 @@ def validate_json_format(value: Any, format_schema: Union[str, dict, None]) -> O # application level (see module header). HOMIE_EMPTY_STRING_PAYLOAD = "\x00" +# The color formats a `color` property's $format may list (convention.md:393). +_COLOR_FORMATS = frozenset({"rgb", "hsv", "xyz"}) + +# The Homie 5 core $description fields (convention.md:212-222). description_extras +# may add extension fields, never one of these. +_DESCRIPTION_CORE_FIELDS = frozenset( + {"homie", "version", "nodes", "name", "type", "children", "root", "parent", "extensions"} +) + @dataclass(frozen=True) class JsonFieldConstraint: @@ -351,6 +384,383 @@ class PropertyDatatype(StrEnum): JSON = "json" +# ── Payload encoding and validation (convention.md:74-150) ────────────────── +# +# Each encoder takes a Python value and returns ``(payload, None)`` for a valid +# wire payload, or ``(None, problem)`` when the datatype cannot represent the +# value. The device refuses to publish a refused value; the /set path uses the +# same grammar to reject a command. + +_INT64_MIN = -(2**63) +_INT64_MAX = 2**63 - 1 +# re.ASCII: a payload's digits are 0-9 only (convention.md:89, :97), where a bare +# \d also matches every other Unicode decimal digit. +_INTEGER_PAYLOAD = re.compile(r"-?\d+", re.ASCII) +_FLOAT_PAYLOAD = re.compile(r"-?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE]-?\d+)?", re.ASCII) +_DURATION_PAYLOAD = re.compile(r"PT(?:\d+H)?(?:\d+M)?(?:\d+(?:\.\d+)?S)?", re.ASCII) +# ISO 8601 (convention.md:132) in its calendar, ordinal and week date forms, +# extended or basic, with an optional time of day and zone designator. +_DATETIME_PAYLOAD = re.compile( + r""" + (?P\d{4}) + (?: + -(?P\d{2})(?:-(?P\d{2}))? + | (?P\d{2})(?P\d{2}) + | -?(?P\d{3}) + | -?W(?P\d{2})(?:-?(?P\d))? + )? + (?: + T(?P\d{2})(?::?(?P\d{2})(?::?(?P\d{2}))?)?(?:[.,]\d+)? + (?:Z|[+-](?P\d{2})(?::?(?P\d{2}))?)? + )? + """, + re.ASCII | re.VERBOSE, +) +_FLOAT_MAX_DECIMAL = decimal.Decimal(sys.float_info.max) +_COLOR_RANGES = {"rgb": (255.0, 255.0, 255.0), "hsv": (360.0, 100.0, 100.0), "xyz": (1.0, 1.0)} + + +def _reject_json_constant(name: str) -> Any: + raise ValueError(f"{name} is not JSON") + + +def _format_float(number: float) -> Optional[str]: + """A finite float in the convention's grammar: no `+` in the exponent (:97).""" + if not math.isfinite(number): + return None + return repr(float(number)).replace("e+", "e") + + +def _encode_integer(value: Any) -> tuple: + if isinstance(value, bool): + return None, "booleanOnNumeric" + if isinstance(value, str): + if not _INTEGER_PAYLOAD.fullmatch(value): + return None, "integerGrammar" + number = int(value) + elif isinstance(value, numbers.Integral): + number = int(value) + elif isinstance(value, numbers.Real): + if not math.isfinite(value) or not float(value).is_integer(): + return None, "integerNotWhole" + number = int(value) + else: + return None, "integerType" + if not _INT64_MIN <= number <= _INT64_MAX: + return None, "integerOutOf64Bit" + return (value if isinstance(value, str) else str(number)), None + + +def _encode_float(value: Any) -> tuple: + if isinstance(value, bool): + return None, "booleanOnNumeric" + if isinstance(value, str): + if not _FLOAT_PAYLOAD.fullmatch(value) or not math.isfinite(float(value)): + return None, "floatGrammar" + return value, None + if isinstance(value, decimal.Decimal): + if not value.is_finite(): + return None, "floatNotFinite" + if abs(value) > _FLOAT_MAX_DECIMAL: + return None, "floatOutOfRange" + return str(value).replace("E+", "E"), None + if isinstance(value, numbers.Integral): + # Float range (convention.md:95-96). + if abs(int(value)) > sys.float_info.max: + return None, "floatOutOfRange" + return str(int(value)), None + if isinstance(value, numbers.Real): + # A non-float Real (numpy's float32, say) keeps its own shortest text when + # that is already a payload: repr(float()) would expand float32 0.1 to + # 0.10000000149011612. + if not isinstance(value, float): + text = str(value) + if _FLOAT_PAYLOAD.fullmatch(text) and math.isfinite(float(text)): + return text, None + try: + payload = _format_float(float(value)) + except OverflowError: + return None, "floatOutOfRange" + return (payload, None) if payload is not None else (None, "floatNotFinite") + return None, "floatType" + + +def _encode_boolean(value: Any) -> tuple: + if not isinstance(value, bool): + return None, "booleanType" + return ("true" if value else "false"), None + + +def _enum_values(format_: Any) -> Optional[list]: + return format_.split(",") if isinstance(format_, str) and format_ else None + + +def _encode_enum(value: Any, format_: Any) -> tuple: + payload = str(value.value) if isinstance(value, Enum) else str(value) + if payload == "": + return None, "enumEmpty" + allowed = _enum_values(format_) + if allowed is not None and payload not in allowed: + return None, "enumNotInFormat" + return payload, None + + +def _check_color_payload(payload: str, format_: Any) -> Optional[str]: + parts = payload.split(",") + kind = parts[0] + ranges = _COLOR_RANGES.get(kind) + if ranges is None: + return "colorType" + allowed = _enum_values(format_) + if allowed is not None and kind not in allowed: + return "colorTypeNotInFormat" + if len(parts) != len(ranges) + 1: + return "colorComponentCount" + for text, upper in zip(parts[1:], ranges): + if not _FLOAT_PAYLOAD.fullmatch(text): + return "colorComponentGrammar" + if not 0.0 <= float(text) <= upper: + return "colorComponentRange" + return None + + +def _encode_color(value: Any, format_: Any) -> tuple: + if isinstance(value, str): + problem = _check_color_payload(value, format_) + return (None, problem) if problem else (value, None) + if isinstance(value, (tuple, list)): + # Components alone: encode them in the property's preferred (first) color + # format (convention.md:393). + allowed = _enum_values(format_) + if not allowed: + return None, "colorFormatMissing" + components = [] + for component in value: + text, problem = _encode_float(component) + if problem: + return None, "colorComponentGrammar" + components.append(text) + payload = ",".join([allowed[0], *components]) + problem = _check_color_payload(payload, format_) + return (None, problem) if problem else (payload, None) + return None, "colorType" + + +def _check_datetime_payload(text: str) -> Optional[str]: + """None if ``text`` is an ISO 8601 date or date-time (convention.md:132), else a problem.""" + match = _DATETIME_PAYLOAD.fullmatch(text) + if not match: + return "datetimeGrammar" + fields = {k: int(v) for k, v in match.groupdict().items() if v is not None} + if "hour" in fields and not fields.keys() & {"day", "bday", "ordinal", "weekday"}: + return "datetimeGrammar" # a time of day needs a complete date + try: + if "month" in fields or "bmonth" in fields: + datetime.date( + fields["year"], fields.get("month", fields.get("bmonth")), fields.get("day", fields.get("bday", 1)) + ) + except ValueError: + return "datetimeRange" + limits = {"ordinal": (1, 366), "week": (1, 53), "weekday": (1, 7), "hour": (0, 24), "minute": (0, 59)} + limits.update({"second": (0, 60), "zhour": (0, 23), "zminute": (0, 59)}) + for name, (low, high) in limits.items(): + if name in fields and not low <= fields[name] <= high: + return "datetimeRange" + return None + + +def _encode_datetime(value: Any) -> tuple: + if isinstance(value, (datetime.datetime, datetime.date)): + # ISO 8601 extended form, `T` separator (convention.md:132). + return value.isoformat(), None + if isinstance(value, str): + problem = _check_datetime_payload(value) + return (None, problem) if problem else (value, None) + return None, "datetimeType" + + +def _encode_duration(value: Any) -> tuple: + if isinstance(value, datetime.timedelta): + if value < datetime.timedelta(0): + return None, "durationNegative" + # PTxHxMxS (convention.md:137-144): days fold into hours, which the + # format has no larger unit than. + hours, rest = divmod(value.days * 86400 + value.seconds, 3600) + minutes, seconds = divmod(rest, 60) + payload = "PT" + if hours: + payload += f"{hours}H" + if minutes: + payload += f"{minutes}M" + if value.microseconds: + payload += f"{seconds}.{value.microseconds:06d}".rstrip("0") + "S" + elif seconds or payload == "PT": + payload += f"{seconds}S" + return payload, None + if isinstance(value, str): + if value == "PT" or not _DURATION_PAYLOAD.fullmatch(value): + return None, "durationGrammar" + return value, None + return None, "durationType" + + +def _decode_json_container(text: str) -> tuple: + """Parse a json payload: an array or object, never NaN/Infinity (convention.md:150).""" + try: + decoded = json.loads(text, parse_constant=_reject_json_constant) + except (ValueError, TypeError): + return None, "jsonInvalid" + if not isinstance(decoded, (list, dict)): + return None, "jsonNotArrayOrObject" + return decoded, None + + +def _encode_json(value: Any) -> tuple: + if isinstance(value, str): + _, problem = _decode_json_container(value) + # A valid JSON string passes through unchanged rather than re-encoded. + return (None, problem) if problem else (value, None) + if not isinstance(value, (list, tuple, dict)): + return None, "jsonNotArrayOrObject" + try: + return json.dumps(value, allow_nan=False), None + except (TypeError, ValueError): + return None, "jsonInvalid" + + +def _encode_property_value(value: Any, datatype: Any, format_: Any = None) -> tuple: + """Encode ``value`` as a Homie 5 payload for ``datatype``. + + Returns ``(payload, None)``, or ``(None, problem)`` when ``datatype`` cannot + represent ``value`` (convention.md:78). ``problem`` is a short camelCase reason. + The payload is the value's text; the ``0x00`` empty-string encoding is applied + by the publisher, and only for ``string`` (convention.md:65-67). + """ + if datatype == PropertyDatatype.INTEGER: + return _encode_integer(value) + if datatype == PropertyDatatype.FLOAT: + return _encode_float(value) + if datatype == PropertyDatatype.BOOLEAN: + return _encode_boolean(value) + if datatype == PropertyDatatype.ENUM: + return _encode_enum(value, format_) + if datatype == PropertyDatatype.COLOR: + return _encode_color(value, format_) + if datatype == PropertyDatatype.DATETIME: + return _encode_datetime(value) + if datatype == PropertyDatatype.DURATION: + return _encode_duration(value) + if datatype == PropertyDatatype.JSON: + return _encode_json(value) + # string, and a missing or unknown datatype (already warned about). + return (str(value.value) if isinstance(value, Enum) else str(value)), None + + +def _numeric_format(format_: Any) -> tuple: + """``(min, max, step)`` as Fractions (each may be None) from ``[min]:[max][:step]``. + + A format that does not parse imposes no constraint. A step that is not > 0 is + ignored (convention.md:390-391). + """ + if not isinstance(format_, str) or not format_: + return None, None, None + parts = format_.split(":") + if len(parts) not in (2, 3): + return None, None, None + + def bound(text: str) -> Optional[Fraction]: + if not _FLOAT_PAYLOAD.fullmatch(text): + return None + return Fraction(text) + + low, high = bound(parts[0]), bound(parts[1]) + step = bound(parts[2]) if len(parts) == 3 else None + if step is not None and step <= 0: + step = None + return low, high, step + + +def _round_and_range_check(number: Fraction, format_: Any, current: Any) -> tuple: + """Round to the format's step, then check min/max (convention.md:397-413). + + The step base is ``min``, else ``max``, else the property's current value; + with none of them there is nothing to round against and the value stands. + Rounding is ``floor(x + 0.5)`` so it always goes up, as :408-413 recommend. + Returns ``(rounded, problem)``. + """ + low, high, step = _numeric_format(format_) + if step is not None: + base = low if low is not None else high + if base is None and isinstance(current, numbers.Real) and not isinstance(current, bool): + if math.isfinite(current): + base = Fraction(current) + if base is not None: + number = math.floor((number - base) / step + Fraction(1, 2)) * step + base + if low is not None and number < low: + return None, "belowMin" + if high is not None and number > high: + return None, "aboveMax" + return number, None + + +def _parse_set_payload(text: str, datatype: Any, format_: Any, current: Any) -> tuple: + """Validate an inbound ``/set`` payload for ``datatype``. + + Returns ``(value, None)`` with the value to hand the set callback, or + ``(None, problem)``. The value is the payload text for every datatype but + ``json`` (the decoded array or object) and ``string`` (with ``0x00`` decoded + to ``""``). A number that step rounding changed is handed over in its + rounded payload form. + """ + if datatype == PropertyDatatype.JSON: + return _decode_json_container(text) + if datatype == PropertyDatatype.INTEGER: + if not _INTEGER_PAYLOAD.fullmatch(text) or not _INT64_MIN <= int(text) <= _INT64_MAX: + return None, "integerGrammar" + rounded, problem = _round_and_range_check(Fraction(int(text)), format_, current) + if problem: + return None, problem + if rounded.denominator != 1: + return None, "integerStepNotWhole" + return (text if rounded == int(text) else str(int(rounded))), None + if datatype == PropertyDatatype.FLOAT: + if not _FLOAT_PAYLOAD.fullmatch(text) or not math.isfinite(float(text)): + return None, "floatGrammar" + number = Fraction(text) + rounded, problem = _round_and_range_check(number, format_, current) + if problem: + return None, problem + return (text if rounded == number else _format_float(float(rounded))), None + if datatype == PropertyDatatype.BOOLEAN: + return (text, None) if text in ("true", "false") else (None, "booleanGrammar") + if datatype == PropertyDatatype.ENUM: + allowed = _enum_values(format_) + if text == "" or (allowed is not None and text not in allowed): + return None, "enumNotInFormat" + return text, None + if datatype == PropertyDatatype.COLOR: + problem = _check_color_payload(text, format_) + return (None, problem) if problem else (text, None) + if datatype == PropertyDatatype.DURATION: + return _encode_duration(text) + if datatype == PropertyDatatype.DATETIME: + return _encode_datetime(text) + # string, and a missing or unknown datatype (already warned about). + return decode_empty_string(text), None + + +def _subscribe_supports_with_retain(mqttc: Any) -> bool: + """Whether ``mqttc.subscribe`` takes ``with_retain`` (ebus-mqtt-client with #24). + + Only a parameter of that name counts: a transport taking ``**kwargs`` may not + deliver the flag. + """ + try: + return "with_retain" in inspect.signature(mqttc.subscribe).parameters + except (TypeError, ValueError): + return False + + def sanitize_homie_id(value: Optional[str]) -> str: """Coerce an arbitrary string to a Homie-legal id segment (a-z, 0-9, -). @@ -509,15 +919,23 @@ def __init__( self._name = id self._datatype = datatype self._format = format - self._settable = settable + self._settable = bool(settable) # Don't assign set_callback unless this property is settable if settable: self._set_callback = set_callback else: self._set_callback = None - self._retained = retained + if retained is None: + # `retained` is a non-nullable boolean in $description (convention.md:343). + # The publish path always treated None as non-retained, so keep that. + _warn_invalid_once(self, "retainedNull", f"reason=propertyRetainedNull,propertyID={id},treatedAs=false") + self._retained = bool(retained) self._unit = unit self._supports_target = supports_target + if supports_target: + # $target is not implemented (publish_target_value() is a stub), so a + # property must never use it (convention.md:333). + logger.warning(f"reason=propertySupportsTargetInert,propertyID={id},hint=$target is not implemented") self._node = node self._device = device self.async_loop = async_loop @@ -569,6 +987,43 @@ def __init__( self._initial_value_was_none = value is None # Check for skip_initial_publish flag from dict self._skip_initial_publish = from_dict.get("skip_initial_publish", False) if from_dict else False + self._check_description_fields() + + def _check_description_fields(self) -> None: + """Warn (once per problem) about $description fields Homie 5 rejects. + + ``datatype`` is required (convention.md:340); ``enum`` and ``color`` require + a ``format``; an enum format lists at least one value, none empty and none + duplicated (:392); a color format lists only ``rgb``, ``hsv`` and ``xyz`` + (:393). + """ + pid = self._id + datatype = self._datatype + if datatype is None: + _warn_invalid_once(self, "datatypeMissing", f"reason=propertyDatatypeMissing,propertyID={pid}") + return + if datatype not in PropertyDatatype._value2member_map_: + _warn_invalid_once( + self, "datatypeUnknown", f"reason=propertyDatatypeUnknown,propertyID={pid},datatype={datatype}" + ) + return + fmt = self._format + if datatype in (PropertyDatatype.ENUM, PropertyDatatype.COLOR) and not fmt: + _warn_invalid_once( + self, "formatMissing", f"reason=propertyFormatMissing,propertyID={pid},datatype={datatype}" + ) + return + if datatype == PropertyDatatype.ENUM and isinstance(fmt, str): + values = fmt.split(",") + if "" in values: + _warn_invalid_once(self, "enumEmptyValue", f"reason=propertyEnumFormatEmptyValue,propertyID={pid}") + if len(set(values)) != len(values): + _warn_invalid_once(self, "enumDuplicate", f"reason=propertyEnumFormatDuplicate,propertyID={pid}") + if datatype == PropertyDatatype.COLOR and isinstance(fmt, str): + if not set(fmt.split(",")) <= _COLOR_FORMATS: + _warn_invalid_once( + self, "colorFormatInvalid", f"reason=propertyColorFormatInvalid,propertyID={pid},format={fmt}" + ) def as_dict(self) -> dict: return { @@ -708,42 +1163,33 @@ def set_format(self, new_format: Optional[str]) -> None: ``_format`` because no setter existed (SDK-6do.2). """ self._format = new_format + self._check_description_fields() def coerced_value(self) -> Optional[str]: """ - Returns the property's value (potentially rounded), as a string. - Returns None if the value is invalid or cannot be coerced. + Returns the property's value (potentially rounded) as its Homie 5 payload + text, or None if the value is None or the datatype cannot represent it + (convention.md:78-150). A refusal is logged at warning. + + datetime/date values encode as ISO 8601, timedelta as ``PTxHxMxS``, a + color tuple as the first ``$format`` color type plus its components, and a + json value must be an array or object (a JSON string of one passes through + unchanged). The ``0x00`` empty-string encoding is not applied here. """ + payload, problem = self._encode() + if problem is not None: + logger.warning( + f"reason=propertyValueRefused,propertyId={self._id},datatype={self._datatype}," + f"problem={problem},value={self._value!r}" + ) + return payload + + def _encode(self) -> tuple: + """``(payload, problem)`` for the current value; both None for a None value.""" property_value = self.value() if property_value is None: - return None - - # A json-datatype property's wire payload MUST be serialized JSON text - # (Homie 5 §JSON), the same serialization used for $description. Mirror - # the inbound /set path (json.loads); do NOT fall through to str(), which - # emits Python repr (single quotes) and is invalid JSON. An already-valid - # JSON string is passed through unchanged so we don't double-encode it. - if self.is_json_datatype(): - if isinstance(property_value, str): - return property_value - try: - return json.dumps(property_value) - except (TypeError, ValueError) as e: - logger.warning(f"reason=coercedValueInvalidJson,propertyId={self._id},value={property_value},error={e}") - return None - - property_type = self.datatype() - if property_type == PropertyDatatype.BOOLEAN: - if not isinstance(property_value, bool): - logger.warning(f"reason=coercedValueInvalidBoolean,propertyId={self._id},value={property_value}") - return None - return str(property_value).lower() - - # For enum values, use .value to get the underlying value - if isinstance(property_value, Enum): - return str(property_value.value) - - return str(property_value) + return None, None + return _encode_property_value(property_value, self._datatype, self._format) def id(self) -> str: """ @@ -845,6 +1291,7 @@ def publish_target_value(self, payload) -> None: """ The $target attribute must either be used for every value update (including the initial one), or it must never be used. TODO: Currently unimplemented, TBD how $target gets set on initial property value set... + Nothing in the SDK calls this; ``supports_target=True`` is inert. """ logger.info(f"reason=propertyPublishTargetValue,propertyID={self._id},value={payload}") logger.warning(f"reason=propertyPublishTargetValueNotImplemented,propertyID={self._id},value={payload}") @@ -909,12 +1356,14 @@ def publish_value(self, *, force: bool = False) -> bool: try: value = self.coerced_value() if value is None: - logger.warning( - f"reason=propertyPublishValueCoercionFailed,propertyID={self._id},rawValue={self._value}" - ) + # Refused (logged by coerced_value): publish nothing rather than a + # payload the datatype does not allow (convention.md:78). return False # Encode an empty-string value as a single 0x00 byte so the broker - # does not mistake it for a zero-length "clear retained" payload. + # does not mistake it for a zero-length "clear retained" payload. Only + # for string (convention.md:65-67); another datatype never encodes to + # "" because its encoder refuses an empty value. A missing datatype was + # warned about at construction and keeps the string behavior. payload = encode_empty_string(value) # GH #50: skip a republish whose final wire payload is byte-identical to # the one already sitting on this topic. Compared AFTER coercion and @@ -925,13 +1374,14 @@ def publish_value(self, *, force: bool = False) -> bool: # RETAINED only. The broker stores nothing for an event property, so an # identical consecutive payload there is a second real event and dropping # it would lose information rather than save a redundant write. - # Truthiness rather than `is True`: retained is Optional[bool] and may be - # None, which the publish call below already treats as non-retained. if not force and self.retained() and self._ever_published and self._last_published == (topic, payload): logger.debug(f"reason=propertyPublishValueUnchanged,propertyID={self._id},topic={topic}") return True logger.debug(f"reason=propertyPublishValue,value={value},topic={topic},retained={self.retained()}") - mqttc.publish(topic, payload, retain=self.retained(), qos=self._qos) + # A non-retained (event) property publishes non-retained at QoS 0 + # (convention.md:50, :695): an event arrives now or not at all. + qos = self._qos if self.retained() else 0 + mqttc.publish(topic, payload, retain=self.retained(), qos=qos) self._ever_published = True # FIX: Mark as published # Memoize only after publish() returns, inside the try: a transport that # raises must not leave a memo claiming the broker holds a payload it @@ -974,6 +1424,13 @@ def clear_value(self) -> bool: if not self._ever_published: logger.info(f"reason=propertySkipClearNeverPublished,propertyID={self._id}") return True + if not self.retained(): + # The broker stores nothing for an event property, so there is nothing + # to retract; a retained empty message on an event topic would deliver + # a zero-length "event" to every live subscriber instead. + self._ever_published = False + self._last_published = None + return True mqttc = self.get_mqtt_client() # See publish_value: gate on connectivity so an injected (caller-driven) @@ -1047,24 +1504,44 @@ def description(self) -> dict: logger.debug(f"reason=propertyDescriptionEntered,id={self._id}") property = dict() property["name"] = self._name - property["datatype"] = self.datatype() + if self._datatype is not None: + # Required (convention.md:340); a missing datatype was warned about at + # construction, and null is not a valid value for it. + property["datatype"] = self.datatype() if self._format: - property["format"] = self.format() + fmt = self.format() + # A json format is a JSONschema carried as a string, NOT a nested + # object (convention.md:395). + property["format"] = json.dumps(fmt) if isinstance(fmt, dict) else fmt if self._settable: - property["settable"] = self._settable + property["settable"] = True if not self._retained: - property["retained"] = self._retained + property["retained"] = False if self._unit: property["unit"] = self._unit return property - def _settable_callback(self, topic: str, payload: Union[bytes, bytearray]) -> None: + def _settable_callback(self, topic: str, payload: Union[bytes, bytearray], retained: Optional[bool] = None) -> None: """ For each settable property, there is a property/set topic that can be published to This is the callback for the subscription to each such property/set topic Examples: [homieDomain]/[homieVerson]/[deviceID]/[nodeID]/mode/set [homieDomain]/[homieVerion]/[deviceID]/[nodeID]/setpoint/set + + ``retained`` is the MQTT retain flag, delivered when the transport supports + ``subscribe(..., with_retain=True)``. A retained ``/set`` is a stale command + the broker replays at subscribe time (controllers publish ``/set`` + non-retained only, convention.md:48, :483), so it is ignored. + + The payload is validated for the datatype before the callback runs, and an + invalid one is dropped with a ``reason=propertySetRejected`` warning: json + must be an array or object (and match a ``$format`` JSONschema when one can + be checked), integer and float must follow the payload grammar, a number is + rounded to the format's step and then checked against min/max + (convention.md:399), an enum must be in ``$format``, a boolean must be + ``true`` or ``false``. The ``0x00`` empty-string decoding applies to string + properties only. """ logger.debug(f"reason=propertySetCallback,topic={topic}") try: @@ -1085,6 +1562,9 @@ def _settable_callback(self, topic: str, payload: Union[bytes, bytearray]) -> No ): logger.debug(f"reason=nodeSetCallbackInvalidTopic,topic={topic}") return + if retained: + logger.warning(f"reason=propertySetRetainedIgnored,propertyID={property_id},topic={topic}") + return # It is possible that we have a valid property/set set_callback = self.get_set_callback() if not self.settable(): @@ -1094,28 +1574,26 @@ def _settable_callback(self, topic: str, payload: Union[bytes, bytearray]) -> No logger.info(f"reason=propertySetCallbackPropertyNoSetCallback,propertyID={property_id}") return try: - decoded_payload = payload.decode("utf-8") # do we need to str() this? - # Homie 5: a single 0x00 byte on /set denotes an empty-string value. - decoded_payload = decode_empty_string(decoded_payload) - if self.is_json_datatype(): - payload = json.loads(decoded_payload) + decoded_payload = payload.decode("utf-8") + value, problem = _parse_set_payload(decoded_payload, self._datatype, self._format, self._value) + if problem is None and self.is_json_datatype(): # Validate the decoded command against the property's $format - # JSONSchema (its advertised control surface). Reject an invalid - # command rather than acting on it. Graceful: skipped if there is - # no $format or the jsonschema package is not installed. - error = validate_json_format(payload, self._format) + # JSONSchema (its advertised control surface). Graceful: skipped if + # there is no $format or the jsonschema package is not installed. + error = validate_json_format(value, self._format) if error is not None: - logger.warning(f"reason=propertySetRejectedSchemaInvalid,propertyID={property_id},error={error}") - return - else: - payload = decoded_payload + problem = f"schemaInvalid:{error}" + if problem is not None: + logger.warning( + f"reason=propertySetRejected,propertyID={property_id},datatype={self._datatype}," + f"problem={problem},payload={decoded_payload!r}" + ) + return + payload = value # We have the payload logger.debug( f"reason=propertySetCallbackValue,propertyID={property_id},payload={payload},callback={set_callback}" ) - if self.supports_target(): - # Property supports_target, publish that! - self.publish_target_value(payload) # Call the property's set_callback function # Run the callback: a sync callback runs inline here (on the transport's # network thread, as before); an async (coroutine) callback is scheduled onto @@ -1169,7 +1647,12 @@ def set_subscribe(self) -> None: return topic = f"{self._homie_domain()}/{EBUS_HOMIE_VERSION_MAJOR}/{device_id}/{node_id}/{self._id}/set" try: - mqttc.subscribe(topic, param=partial(self._settable_callback), qos=self._qos) + if _subscribe_supports_with_retain(mqttc): + # Deliver the retain flag so a stale retained /set is ignored. + mqttc.subscribe(topic, param=partial(self._settable_callback), qos=self._qos, with_retain=True) + else: + logger.debug(f"reason=propertySetSubscribeNoRetainFlag,id={self._id},hint=retained /set not detected") + mqttc.subscribe(topic, param=partial(self._settable_callback), qos=self._qos) except Exception as e: logger.warning(f"reason=propertySetSubscribeSubscribeException,e={e}") # Start the MQTT client loop() thread @@ -1285,17 +1768,22 @@ def add_property(self, property: Property) -> Property: property.async_loop = self._device._async_loop # Note set_subscribe() checks if property is settable... property.set_subscribe() - # Add property to dictionary BEFORE publishing description + # Outside a transition a ready device goes to init first, so the value lands + # before the $description that names it and before ready (convention.md:279). + # A node not yet attached to its device, or a re-declaration with identical + # content, leaves the $description unchanged and publishes only the value. self._properties.update({property.id(): property}) - self.device().publish_description() - # force: announcing a property is a structural republish, so its value must - # land regardless of the GH #50 skip. A fresh property would pass the gate - # anyway (it has never published), and so would one re-added after - # delete_property() (clear_value() resets both gate conjuncts). What force - # actually covers is the property whose retained topic was deleted behind its - # back -- clear_retained_topic(), or an operator wiping the broker -- where - # the memo still claims the broker holds this payload and it does not. - property.publish_value(force=True) + with self.device()._reconfiguration(): + # force: announcing a property is a structural republish, so its value must + # land regardless of the GH #50 skip. A fresh property would pass the gate + # anyway (it has never published), and so would one re-added after + # delete_property() (clear_value() resets both gate conjuncts). What force + # actually covers is the property whose retained topic was deleted behind its + # back -- clear_retained_topic(), or an operator wiping the broker -- where + # the memo still claims the broker holds this payload and it does not. + # An event property is not replayed by a structural change (:695). + if property.retained(): + property.publish_value(force=True) return property def add_property_from_dict(self, property_dict: dict) -> Property: @@ -1338,13 +1826,16 @@ def delete_property(self, property_id: str) -> bool: logger.warning(f"reason=nodeDeletePropertyNotFound,nodeId={self._id},propertyId={property_id}") return False property = self._properties[property_id] - property.clear_value() - del self._properties[property_id] - # Delete from the dict BEFORE republishing, so the new $description - # reflects the removal (add_property() has the same ordering rule). device = self.device() if device: - device.publish_description() + # The description that drops the property is published when the + # reconfiguration scope closes, after the removal. + del self._properties[property_id] + with device._reconfiguration(): + property.clear_value() + else: + property.clear_value() + del self._properties[property_id] logger.info(f"reason=nodeDeletedProperty,nodeId={self._id},propertyId={property_id}") return True @@ -1378,7 +1869,9 @@ def description(self) -> dict: logger.debug(f"reason=nodeDescriptionEntered,id={self._id}") description = dict() description["name"] = self._name - description["type"] = self._type + if self._type is not None: + # Optional and not nullable (convention.md:305): omitted when unset. + description["type"] = self._type properties = dict() properties_snapshot = dict(self._properties) for property_id, attributes in properties_snapshot.items(): @@ -1400,6 +1893,11 @@ def publish(self, *, force: bool = True) -> None: logger.debug(f"reason=nodePublish,nodeId={node_id},propertyCount={property_count}") # Use list() to create a shallow copy, preventing crash if dict changes during iteration for property_id, property in list(self._properties.items()): + if not property.retained(): + # A republish walk restores the broker's retained store. An event + # property has nothing there, and replaying its last event would + # report it again (convention.md:695). + continue logger.debug(f"reason=nodePublishProperty,nodeId={node_id},propertyId={property_id}") # Best-effort per property. Property.publish_value() reaches the MQTT # client directly and does not wrap it, so an injected transport that @@ -1467,6 +1965,41 @@ def _dispatch_disconnect(callback: Optional[Callable[[bool], None]], rc, source: logger.exception(f"reason=onDisconnectCallbackException,source={source}") +def _drop_connection_without_disconnect(client: Any, *, timeout: float) -> None: + """End an owned client's connection WITHOUT an MQTT DISCONNECT packet. + + The broker then treats the connection as badly disconnected and publishes the + Last Will. ``Device.stop(announce=False)`` uses this so a teardown that announced + nothing is a bad disconnect (convention.md:281, :284) rather than a clean one. + + ebus-mqtt-client has no unclean-stop API, so this stops the paho network loop + (which sends nothing) and shuts the socket down underneath it. The client's own + ``stop()`` runs afterwards to release its resources; its DISCONNECT then has no + live connection to reach. Bounded by ``timeout`` like ``MqttClient.stop()``. + """ + paho = getattr(client, "mqttc", None) + if paho is None or not callable(getattr(paho, "socket", None)): + logger.warning("reason=deviceDropConnectionUnsupported,hint=the will may not fire") + return + + def _drop() -> None: + with contextlib.suppress(Exception): + paho.loop_stop() + sock = paho.socket() + if sock is None: + return + with contextlib.suppress(Exception): + sock.shutdown(socket.SHUT_RDWR) + with contextlib.suppress(Exception): + sock.close() + + worker = threading.Thread(target=_drop, name="ebus-sdk-drop-connection", daemon=True) + worker.start() + worker.join(timeout) + if worker.is_alive(): + logger.warning(f"reason=deviceDropConnectionTimeout,timeout={timeout}") + + class Device: """ Object representing a Homie MQTT Device @@ -1654,14 +2187,28 @@ def __init__( # description fields (convention §Forward compatibility), so these are # safe; core fields always take precedence over an extra of the same key. self._description_extras = dict(description_extras) if description_extras else {} + for key in sorted(self._description_extras.keys() & _DESCRIPTION_CORE_FIELDS): + # A core field is the SDK's to compute: an extra could otherwise add + # `root`/`parent` to a root device (convention.md:220) or a `type` the + # SDK omits. Dropped, with a warning per the invalid-input policy. + del self._description_extras[key] + _warn_invalid_once( + self, f"extrasCoreField:{key}", f"reason=deviceDescriptionExtrasCoreFieldDropped,id={id},field={key}" + ) # Counter of how many state_transition() / delete() scopes are currently active # on this device. >0 means "a transition is in progress" — suppresses child-induced # parent flaps and makes nested state_transition()s reentrant (only the outermost # entry/exit publishes INIT/READY). Init→ready transitions force every controller # in the wild to resync, so emitting only the minimum is a correctness concern. self._transition_depth = 0 - # SDK-n83: hash of the last $description we actually published, with the - # always-fresh `version` timestamp removed. publish_description() uses it + # Set by delete(): the device no longer exists on the broker, so nothing may + # republish its $state (stop(), a reconnect's refresh_tree(), the will). + self._deleted = False + # Alerts this device has raised and not cleared: alert id -> message. + # Tracked so a reconnect republishes them and delete() clears them. + self._alerts: dict = {} + # SDK-n83: hash of the last $description we actually published, with + # `version` removed. publish_description() uses it # to skip a republish whose content has not changed (saves the ~KB # payload and the gratuitous INIT→READY flap). Maintained in publish(). self._last_description_content_hash = None @@ -1862,71 +2409,115 @@ def is_connected(self) -> bool: def stop(self, *, announce: bool = True, flush_timeout: float = 1.0, stop_timeout: float = 2.0) -> None: """Gracefully and promptly tear down this device tree's MQTT connection. - Publishes a final ``$state=disconnected`` for the root (best-effort) then - stops the root's MQTT client. This is a TREE-level teardown: children - share the root's connection, so a call on any device stops the whole - tree; per the Homie 5 effective-state rule, the root going - ``disconnected`` covers every descendant. + Publishes a final ``$state=disconnected`` for EVERY device in the tree + (best-effort), descendants first and the root last, then stops the root's + MQTT client with a clean disconnect. Homie 5 requires a device to send + ``disconnected`` before it cleanly disconnects (convention.md:280-281), and + children share the root's connection, so each of them disconnects too; the + root-to-child cascade (convention.md:262-267) covers only ``lost``. This is + a TREE-level teardown: a call on any device stops the whole tree. BOUNDED end to end (at most ~``flush_timeout`` + ``stop_timeout``): if the - broker is unreachable the disconnected publish is skipped and ``stop()`` + broker is unreachable the disconnected publishes are skipped and ``stop()`` still returns promptly, so a shutting-down process never stalls on a dead - broker. Because ``MqttClient.stop()`` performs a clean disconnect (which - suppresses the LWT), publishing ``$state=disconnected`` first is what lets - consumers see a clean shutdown rather than a stale ``ready`` or a - badly-disconnected ``lost``. After ``stop()`` this device tree should not - be reused. + broker. The root's publish is the one flushed; it is sent last, so the + children's publishes precede it on the wire. Because ``MqttClient.stop()`` + performs a clean disconnect (which suppresses the LWT), publishing + ``disconnected`` first is what lets consumers see a clean shutdown rather + than a stale ``ready``. After ``stop()`` this device tree should not be + reused. For a bring-your-own-transport root (``mqttc=``) the SDK does not own the - client: ``stop()`` publishes the final ``$state=disconnected`` as a plain - retained message and returns immediately, without flushing and without - closing the client, so it never blocks the caller's loop. The caller - stops and disconnects its own client. ``flush_timeout``/``stop_timeout`` - apply to the owned path only. - - ``announce=False`` tears down without publishing anything, leaving the - retained ``$state`` exactly as it stands. Pair it with ``declare_lost()`` - for a producer that is dying rather than shutting down: that publishes - ``lost`` first, and the default ``announce=True`` would then overwrite it - with ``disconnected``. Unpaired it leaves whatever was published last, - typically a stale ``ready``, and nothing will correct that: the clean - disconnect on the owned path suppresses the LWT (see above). The teardown - itself stays bounded and clean in both modes; only the announcement differs. + client: ``stop()`` publishes the same ``$state=disconnected`` messages as + plain retained messages and returns immediately, without flushing and + without closing the client, so it never blocks the caller's loop. The + caller stops and disconnects its own client. ``flush_timeout`` and + ``stop_timeout`` apply to the owned path only. + + ``announce=False`` publishes nothing and ends the connection WITHOUT a + clean MQTT DISCONNECT, so the broker delivers the Last Will (the root's + ``$state=lost``). Homie 5 forbids a clean disconnect without a preceding + ``disconnected`` (convention.md:281) and defines ``lost`` as a bad + disconnect (convention.md:284), so a silent teardown is a bad disconnect. + Pair it with ``declare_lost()`` for a producer that is dying rather than + shutting down: that publishes ``lost`` while the link is up, and the will + then re-asserts the same retained payload. On a bring-your-own transport + the SDK closes nothing; the caller must drop its connection without a clean + DISCONNECT for the same guarantee. + + After ``delete()`` on the root, ``stop()`` announces nothing in either mode + and always disconnects cleanly: the device no longer exists + (convention.md:288), and a ``disconnected`` or a will-driven ``lost`` would + re-create it as a bare ``$state`` with no ``$description``. """ root = self.root() mqttc = root.mqttc if mqttc is None: _log_missing_client(f"reason=deviceStopNoMqttClient,id={self._id}", by_design=self._transport_free()) return - # Best-effort graceful $state=disconnected. publish_and_flush is bounded - # and returns False (never blocks/raises) when the broker is unreachable, - # so this can't stall shutdown. Note the state move lives inside this branch, - # so announce=False cannot overwrite a $state a caller just declared. - if not announce: + owned = root._owns_client and root._owned_client is not None + if root._deleted: + logger.info(f"reason=deviceStopDeletedTreeSilent,id={root._id}") + if owned and mqttc.is_connected(): + # Nothing to announce, but delete()'s retained clears may still be in + # flight. Re-clearing the root's $state (a no-op deletion) is flushed, + # and with it everything queued before it, ahead of the disconnect. + state_topic = f"{root.homie_domain()}/{EBUS_HOMIE_VERSION_MAJOR}/{root._id}/$state" + root._owned_client.publish_and_flush(state_topic, "", qos=root._qos, retain=True, timeout=flush_timeout) + elif not announce: logger.info(f"reason=deviceStopSilent,id={root._id}") + if owned: + # No disconnected was sent, so the disconnect must not be clean: drop + # the connection and let the broker publish the will. + _drop_connection_without_disconnect(root._owned_client, timeout=stop_timeout) elif mqttc.is_connected(): - root._state = DeviceState.DISCONNECTED - state_topic = f"{root.homie_domain()}/{EBUS_HOMIE_VERSION_MAJOR}/{root._id}/$state" - if root._owns_client and root._owned_client is not None: - flushed = root._owned_client.publish_and_flush( - state_topic, DeviceState.DISCONNECTED.value, qos=root._qos, retain=True, timeout=flush_timeout - ) - logger.info(f"reason=deviceStopDisconnectedPublished,id={root._id},flushed={flushed}") - else: - # Bring-your-own-transport: the caller owns the loop and teardown. - # Publish the final state as a plain retained message (the caller's - # loop delivers it) and return without flushing or closing — both - # publish_and_flush and stop() are owned-only, off the injected - # transport surface, so this never blocks the caller's thread. - mqttc.publish(state_topic, DeviceState.DISCONNECTED.value, retain=True, qos=root._qos) - logger.info(f"reason=deviceStopDisconnectedPublishedInjected,id={root._id}") + # Best-effort graceful $state=disconnected for the whole tree. The state + # moves live inside this branch, so announce=False cannot overwrite a + # $state a caller just declared. + for device in root._tree_post_order(): + device._state = DeviceState.DISCONNECTED + state_topic = f"{root.homie_domain()}/{EBUS_HOMIE_VERSION_MAJOR}/{device._id}/$state" + if device is root and owned: + # publish_and_flush is bounded and returns False (never blocks or + # raises) when the broker is unreachable, so this cannot stall + # shutdown. + flushed = root._owned_client.publish_and_flush( + state_topic, DeviceState.DISCONNECTED.value, qos=root._qos, retain=True, timeout=flush_timeout + ) + logger.info(f"reason=deviceStopDisconnectedPublished,id={root._id},flushed={flushed}") + continue + # A descendant, or any device on an injected transport: a plain + # retained publish. publish_and_flush is owned-only and off the + # injected transport surface, so this never blocks the caller's thread. + try: + mqttc.publish(state_topic, DeviceState.DISCONNECTED.value, retain=True, qos=root._qos) + except Exception as e: + logger.warning(f"reason=deviceStopDisconnectedPublishFailed,id={device._id},e={e}") + continue + if device is root: + logger.info(f"reason=deviceStopDisconnectedPublishedInjected,id={root._id}") else: logger.info(f"reason=deviceStopBrokerUnreachable,id={root._id}") - if root._owns_client and root._owned_client is not None: + if owned: root._owned_client.stop(timeout=stop_timeout) root.mqttc = None root._owned_client = None + def _tree_post_order(self) -> List["Device"]: + """This device and every descendant, each device after all of its descendants.""" + ordered: List[Device] = [] + for child in list(self._children): + ordered.extend(child._tree_post_order()) + ordered.append(self) + return ordered + + def _tree_pre_order(self) -> List["Device"]: + """This device and every descendant, each device before any of its descendants.""" + ordered: List[Device] = [self] + for child in list(self._children): + ordered.extend(child._tree_pre_order()) + return ordered + def declare_lost(self, *, flush_timeout: float = 1.0) -> bool: """Declare this device tree dead: move the ROOT to ``$state=lost`` and publish it. @@ -2001,9 +2592,12 @@ def description(self) -> dict: logger.debug(f"reason=deviceDescriptionEntered,id={self._id}") description = dict() description["homie"] = f"{EBUS_HOMIE_VERSION_MAJOR}.{EBUS_HOMIE_VERSION_MINOR}" - # Version should be changed any time the description document is changed - description["version"] = Device.now_ems() - description["type"] = self._type + # Placeholder keeps `version` second in the document; filled in below from + # the content, which the hash computes without it. + description["version"] = 0 + if self._type is not None: + # Optional and not nullable (convention.md:218): omitted when unset. + description["type"] = self._type description["name"] = self._name nodes_descriptions = dict() nodes_snapshot = dict(self._nodes) @@ -2017,9 +2611,15 @@ def description(self) -> dict: # Required if the parent is NOT the root device. Defaults to the value of the root property. description["parent"] = self._parent.id() description["extensions"] = self._extensions - # Merge extension-defined device attributes, never clobbering a core field. + # Merge extension-defined device attributes (core fields were dropped from + # the extras at construction). for key, value in self._description_extras.items(): description.setdefault(key, value) + # A new version whenever the document changes (convention.md:215), and the + # same version while it does not, so a reconnect republishes an identical + # document: the top 52 bits of the content hash, an integer below 2^53 that + # every JSON consumer reads exactly. + description["version"] = int(self._description_content_hash(description)[:13], 16) return description def set_state(self, state: DeviceState) -> bool: @@ -2054,12 +2654,14 @@ def add_node(self, node: Node) -> Node: if self._async_loop is not None: prop.async_loop = self._async_loop node_id = node.id() + # Outside a transition a ready device goes to init first, so the node's + # values land before the $description that names them and before ready. self._nodes.update({node_id: node}) - # Explicit force (also Node.publish's default): adopting a node is a - # structural republish, so every property lands on the broker under this - # device regardless of the GH #50 unchanged-payload skip. - node.publish(force=True) - self.publish_description() + with self._reconfiguration(): + # Explicit force (also Node.publish's default): adopting a node is a + # structural republish, so every property lands on the broker under this + # device regardless of the GH #50 unchanged-payload skip. + node.publish(force=True) return node def add_node_from_dict(self, node_dict: dict) -> Node: @@ -2090,7 +2692,8 @@ def remove_node(self, node_id: str) -> bool: """ if node_id in self._nodes: self._nodes.pop(node_id, None) - self.publish_description() + with self._reconfiguration(): + pass return True else: return False @@ -2116,20 +2719,19 @@ def delete_node(self, node_id: str) -> bool: logger.warning(f"reason=deviceDeleteNodeNotFound,deviceId={self._id},nodeId={node_id}") return False node = self._nodes[node_id] - # Clear all property topics first - # Note: This explicitly clears each property's retained message from MQTT - # to avoid leaving orphaned topics in the broker - node.clear_all_properties() - # Remove node from device's internal structure + # The description that drops the node is published when the reconfiguration + # scope closes, after its property topics are cleared. del self._nodes[node_id] - # Update device description (which removes the node from the schema) - self.publish_description() + with self._reconfiguration(): + # Clear each published property's retained message so no orphaned + # topics are left in the broker. + node.clear_all_properties() logger.info(f"reason=deviceDeletedNode,deviceId={self._id},nodeId={node_id}") return True def delete_all_from_mqtt(self) -> None: """ - Clear this device's retained property values and $description from the broker. + Clear this device's retained property values, alerts and $description from the broker. A low-level data-cleanup helper: it clears every published property value and the $description topic, but deliberately does NOT touch $state, so on its own it @@ -2167,7 +2769,10 @@ def delete_all_from_mqtt(self) -> None: elif hasattr(prop, "_ever_published") and prop._ever_published: was_published = True - if was_published: + if was_published and not prop.retained(): + # Nothing retained to clear on an event topic (see clear_value). + prop.invalidate_publish_cache() + elif was_published: prop_topic = f"{base_topic}/{node_id}/{prop_id}" try: mqttc.publish(prop_topic, "", retain=True, qos=self._qos) @@ -2186,66 +2791,80 @@ def delete_all_from_mqtt(self) -> None: description_topic = f"{base_topic}/$description" try: mqttc.publish(description_topic, "", retain=True, qos=self._qos) + self.invalidate_description_cache() logger.info(f"reason=deviceClearedDescription,deviceId={self._id},topic={description_topic}") except Exception as e: logger.warning(f"reason=deviceClearDescriptionFailed,deviceId={self._id},error={e}") - # Step 3: Clear internal tracking (no publishing happens here) + # Step 3: Clear raised alerts; a removed device leaves no topics behind + # (convention.md:289). + for alert_id in list(self._alerts): + self.clear_alert(alert_id) + + # Step 4: Clear internal tracking (no publishing happens here) self._nodes.clear() logger.info(f"reason=deviceDeleteAllFromMqttComplete,deviceId={self._id}") def delete(self) -> None: """ - Remove this device from the tree (Homie remove-child protocol). - - On a child: clears the child's retained MQTT data (state, description, - all property values), detaches from parent, then triggers the parent - to republish its $description (without this child in `children`). - Parent's INIT→READY flap is suppressed if the parent is already mid - state_transition (S3: batched remove inside `with parent.state_transition()`). - - On a root: clears all retained data for this device. (Does not stop - the MQTT client — that's the caller's responsibility, after which the - LWT publish covers the whole tree.) Note this REMOVES the device: an absent - retained ``$state`` is the Homie removal signal, so do not follow it with - ``declare_lost()``, which would resurrect the device on the broker as a bare - ``$state=lost`` with no ``$description``. (The root's will is a separate - matter: it is armed on the connection, not on the device, so it still fires - if the process then dies uncleanly.) To retire a tree as dead but still - present, use ``declare_lost()`` plus ``stop(announce=False)`` instead. - - While delete() is running, this device acts as if it were mid - state_transition so descendants' delete()-triggered parent-flap - notifications collapse into nothing — a recursive teardown shouldn't - publish gratuitous INIT/READY on dying devices. + Remove this device, and every descendant, from the broker (Homie removal). + + On a child, in the order convention.md:635-641 gives: the parent is updated + first (``init``, ``$description`` without this child in ``children``, + ``ready``), then this device's topics are cleared starting with its + ``$state``. The parent's update is deferred, as for any structural change, + when the parent is already inside a ``state_transition()``. + + The removed subtree is cleared leaves first, and each device's ``$state`` + is cleared before its other topics: values, alerts and ``$description`` + (convention.md:287-289). No device in the subtree publishes an + ``init``/``ready`` flap on its way out. + + On a root: clears all retained data for the tree. It does not stop the MQTT + client; ``stop()`` afterwards disconnects cleanly and announces nothing, + and a reconnect republishes nothing, since an absent retained ``$state`` is + the removal signal and any later ``$state`` would re-create the device with + no ``$description``. The will is armed on the connection, so it still fires + if the process dies uncleanly before ``stop()``. Do not follow this with + ``declare_lost()``; to retire a tree as dead but still present, use + ``declare_lost()`` plus ``stop(announce=False)`` instead. After delete(), this Device object should not be used further. """ logger.info(f"reason=deviceDelete,deviceId={self._id},isRoot={self._parent is None}") - # Bump transition depth so structural-change notifications from descendants - # we're about to tear down get suppressed — they'd be calling - # _notify_structural_change on a corpse. - self._transition_depth += 1 - try: - # Recursively delete children first so the broker sees a leaves-first cleanup. - for child in list(self._children): - child.delete() - # Clear $state FIRST, per the Homie 5 removal order (convention: clear - # the retained $state and "the device will cease to exist", then clear - # its other retained topics). delete_all_from_mqtt only handles property - # values and $description, so $state is cleared here separately. - base_topic = f"{self.homie_domain()}/{EBUS_HOMIE_VERSION_MAJOR}/{self._id}" - self.clear_retained_topic(f"{base_topic}/$state") - self.delete_all_from_mqtt() - finally: - self._transition_depth -= 1 - if self._parent is not None: - parent = self._parent + subtree = self._tree_pre_order() + for device in subtree: + device._deleted = True + parent = self._parent + if parent is not None: + # Step 1: the parent drops this child from `children` and re-announces. + # The child keeps its _parent link until its topics are cleared: the + # topic domain and the MQTT client are both resolved through the root. parent._children.remove(self) - self._parent = None - # Only fires if parent isn't itself mid-delete or mid state_transition. parent._notify_structural_change() + # Step 2: clear the subtree leaves first, so no device is seen whose parent + # is already gone. Each device's $state goes before its other topics, so it + # ceases to exist first (convention.md:287-289). + for device in reversed(subtree): + base_topic = f"{device.homie_domain()}/{EBUS_HOMIE_VERSION_MAJOR}/{device._id}" + device.clear_retained_topic(f"{base_topic}/$state") + device.delete_all_from_mqtt() + for device in subtree: + device._children = [] + self._parent = None + + def invalidate_description_cache(self) -> None: + """Forget which ``$description`` this device last published. + + ``publish_description()`` skips a document identical to the last one it + published, on the assumption that the broker still holds it. Anything that + deletes the retained ``$description`` behind the device's back must call + this, or the next ``publish_description()`` is skipped and the device stays + without a description. ``clear_retained_topic()`` aimed at this device's + ``$description`` and ``delete_all_from_mqtt()`` call it themselves. + """ + self._last_description_content_hash = None def clear_retained_topic(self, topic_path: str) -> bool: """ @@ -2267,6 +2886,9 @@ def clear_retained_topic(self, topic_path: str) -> bool: try: mqttc.publish(topic_path, "", retain=True, qos=self._qos) logger.info(f"reason=deviceClearedTopic,topic={topic_path}") + if topic_path == f"{self.homie_domain()}/{EBUS_HOMIE_VERSION_MAJOR}/{self._id}/$description": + # The broker no longer holds the document the hash describes. + self.invalidate_description_cache() return True except Exception as e: logger.warning(f"reason=deviceClearTopicException,topic={topic_path},e={e}") @@ -2309,7 +2931,7 @@ def _notify_structural_change(self) -> None: """ Republish this device's $description after a structural change (a child was added or removed). Performs INIT → publish description - → READY unless this device is already inside a state_transition() — + → READY (or back to sleeping) unless this device is already inside a state_transition() — in which case the in-progress transition will publish on exit and we suppress the per-change flap (S1: many children, one parent cycle). @@ -2322,13 +2944,53 @@ def _notify_structural_change(self) -> None: f"reason=deviceStructuralChangeSuppressed,deviceId={self._id},transitionDepth={self._transition_depth}" ) return - if self._state != DeviceState.READY: - self.publish_description(republish=True) + # publish_description() wraps the change in INIT when the device is ready + # or sleeping, and publishes it directly otherwise. + self.publish_description() + + def _description_changed(self) -> bool: + """Whether ``$description`` now differs from the one last published.""" + try: + return self._description_content_hash(self.description()) != self._last_description_content_hash + except Exception: + # publish_description() logs the compose failure. + return True + + @contextlib.contextmanager + def _reconfiguration(self): + """Publish a structural change made outside an explicit ``state_transition()``. + + The caller applies the change to the in-memory tree first, without + publishing, then publishes inside this scope what the change requires (new + values included). + + Homie 5 reconfiguration (convention.md:207, :279, :611): a ``ready`` or + ``sleeping`` device whose ``$description`` the change alters goes to + ``init``, runs the body, publishes the new ``$description``, and only then + returns to the state it was in, so ``ready`` is never announced ahead of the + values it vouches for. A change that leaves the ``$description`` as last + published (a node not attached to the device, an identical re-declaration) + runs the body with no ``init``/``ready`` flap (convention.md:616). Inside an + open transition the change joins it. In any other state (not yet announced, + ``init``, ``disconnected``, ``lost``) the body and the new ``$description`` + are published directly. + """ + if self._transition_depth > 0: + yield + return + prior = self._state + if prior not in (DeviceState.READY, DeviceState.SLEEPING) or not self._description_changed(): + yield + self.publish_description() return - # Steady-state structural change: full INIT → desc → READY cycle. + self._transition_depth += 1 self.set_state(DeviceState.INIT) - self.publish("$description") - self.set_state(DeviceState.READY) + try: + yield + finally: + self._transition_depth -= 1 + self.publish_description() + self.set_state(prior) def state_transition(self) -> StateTransitionContext: """ @@ -2368,6 +3030,10 @@ def refresh_tree(self, *, force: bool = True) -> None: re-raised; callers wanting to detect it should watch for ``reason=deviceRefreshTreeChildFailed`` in the log. """ + if self._deleted: + # delete() removed this device; republishing would re-create it. + logger.info(f"reason=deviceRefreshTreeDeletedSkipped,deviceId={self._id}") + return logger.info( f"reason=deviceRefreshTree,deviceId={self._id}," f"nodeCount={len(self._nodes)},childCount={len(self._children)}" @@ -2383,8 +3049,15 @@ def refresh_tree(self, *, force: bool = True) -> None: # finishes makes the refresh one atomic commit, flipped by one publish. # This narrows a producer-side window; it is NOT a guarantee a consumer # may build on (see doc/consuming-a-homie-tree.md). Message set unchanged. + # A document that differs from the last one published (its memo was + # invalidated, or never stored) is a change, so a ready or sleeping device + # announces init first (convention.md:207); the state publish below ends it. + if self._state in (DeviceState.READY, DeviceState.SLEEPING) and self._description_changed(): + self.publish_state(DeviceState.INIT) self.publish_description(republish=True) self.publish_nodes(force=force) + for alert_id, message in list(self._alerts.items()): + self._publish_alert_topic(alert_id, message) # Snapshot — main thread may construct child devices (which append # to self._children) while this runs on the MQTT loop thread. for child in list(self._children): @@ -2433,17 +3106,17 @@ def publish(self, attribute: str = "", value: Optional[Any] = None) -> None: description = value if value else self.description() payload = json.dumps(description) if description else None elif attribute == "$alert": - topic = base_topic + "$alert" - if value: - payload = value - else: - logger.info(f"reason=devicePublishAlertNoValue,id={self._id}") - return + # An alert lives at $alert/[alert ID] (convention.md:517); a bare + # $alert topic is not one. Use publish_alert(). + logger.warning( + f"reason=devicePublishAlertWithoutId,id={self._id},hint=use publish_alert(alert_id, message)" + ) + return if payload: mqttc.publish(topic, payload, retain=True, qos=self._qos) if attribute == "$description": - # SDK-n83: remember what we just put on the wire (sans the - # version timestamp) so a later unchanged republish no-ops. + # SDK-n83: remember what we just put on the wire (sans + # version) so a later unchanged republish no-ops. # Updated here — the single $description chokepoint — so every # caller (publish_description, _notify_structural_change, # reconnect) keeps the hash current. @@ -2451,6 +3124,75 @@ def publish(self, attribute: str = "", value: Optional[Any] = None) -> None: except Exception as e: logger.exception(f"reason=devicePublishException,id={self._id},attribute={attribute},value={value},e={e}") + def _alert_topic(self, alert_id: str) -> str: + return f"{self.homie_domain()}/{EBUS_HOMIE_VERSION_MAJOR}/{self._id}/$alert/{alert_id}" + + def _publish_alert_topic(self, alert_id: str, message: str) -> bool: + mqttc = self.get_mqtt_client() + if not mqttc: + _log_missing_client( + f"reason=devicePublishAlertNoMqttClient,id={self._id}", by_design=self._transport_free() + ) + return False + try: + mqttc.publish(self._alert_topic(alert_id), message, retain=True, qos=self._qos) + return True + except Exception as e: + logger.warning(f"reason=devicePublishAlertException,id={self._id},alertId={alert_id},e={e}") + return False + + def publish_alert(self, alert_id: str, message: str) -> bool: + """Raise a user-facing alert at ``$alert/[alert_id]`` (convention.md:512-527). + + Retained, at the tree QoS. Raising the same id again replaces its message. + The alert stays until ``clear_alert(alert_id)``; ``delete()`` clears every + alert the device raised, and a reconnect republishes them. + + ``alert_id`` must be a single topic level; one containing ``/``, ``+`` or + ``#``, or an empty one, is refused. An id outside the Homie ID format + (``a-z``, ``0-9``, ``-``) is warned about and still published. ``message`` + must be a non-empty string: an empty payload deletes the topic, so clear an + alert with ``clear_alert()``. Returns True if published. + """ + if not alert_id or not isinstance(alert_id, str) or any(c in alert_id for c in "/+#"): + logger.warning(f"reason=devicePublishAlertInvalidId,id={self._id},alertId={alert_id!r}") + return False + if not re.fullmatch(r"[a-z0-9-]+", alert_id): + _warn_invalid_once( + self, f"alertId:{alert_id}", f"reason=devicePublishAlertIdFormat,id={self._id},alertId={alert_id}" + ) + if not isinstance(message, str) or message == "": + logger.warning(f"reason=devicePublishAlertEmptyMessage,id={self._id},alertId={alert_id}") + return False + self._alerts[alert_id] = message + return self._publish_alert_topic(alert_id, message) + + def clear_alert(self, alert_id: str) -> bool: + """Remove the alert ``alert_id`` by deleting its retained topic (convention.md:519). + + Returns True if the deletion was published. Clearing an alert this device + did not raise still publishes the deletion, so a stale alert left by an + earlier run can be removed. + """ + if not alert_id or not isinstance(alert_id, str) or any(c in alert_id for c in "/+#"): + logger.warning(f"reason=deviceClearAlertInvalidId,id={self._id},alertId={alert_id!r}") + return False + self._alerts.pop(alert_id, None) + mqttc = self.get_mqtt_client() + if not mqttc: + _log_missing_client(f"reason=deviceClearAlertNoMqttClient,id={self._id}", by_design=self._transport_free()) + return False + try: + mqttc.publish(self._alert_topic(alert_id), "", retain=True, qos=self._qos) + return True + except Exception as e: + logger.warning(f"reason=deviceClearAlertException,id={self._id},alertId={alert_id},e={e}") + return False + + def alerts(self) -> dict: + """The alerts this device has raised and not cleared, ``{alert_id: message}``.""" + return dict(self._alerts) + def publish_state(self, state: Optional[DeviceState] = None) -> None: """ Publishes the value of the state argument to the device's $state topic, @@ -2464,51 +3206,59 @@ def publish_state(self, state: Optional[DeviceState] = None) -> None: @staticmethod def _description_content_hash(description: dict) -> str: """ - SHA-256 of a $description dict with the always-fresh `version` timestamp - removed, so two structurally-identical descriptions hash equal even - though description() stamps a new version on every call. + SHA-256 of a $description dict with `version` removed. ``description()`` + derives ``version`` from this hash, and the publish gate compares it to + decide whether the document changed. """ content = {k: v for k, v in description.items() if k != "version"} return hashlib.sha256(json.dumps(content, sort_keys=True).encode()).hexdigest() def publish_description(self, republish: bool = False) -> None: + """Publish this device's ``$description`` if its content changed. + + Homie 5 lets ``$description`` change only while ``$state`` is ``init``, + ``disconnected`` or ``lost`` (convention.md:207). A change made while the + device is ``ready`` or ``sleeping`` is therefore wrapped: ``init``, the new + document, then the state the device was in. + + Unchanged content is skipped. ``version`` is derived from the content + (see ``description()``), so ``republish=True`` (the reconnect path) puts + the same document back on the broker, which is not a change and needs no + ``init``. ``republish=True`` publishes the document without the wrap even + when it changed: ``refresh_tree()`` publishes ``init`` ahead of it and the + device's state after everything the document names. ``republish=True`` is + also exempt from the in-transition defer. + """ # SDK-9ps: while a state_transition() is open, defer interim $description # publishes to the single consolidated publish at _end_state_transition(). # Adding N nodes inside one transition then puts 1 description on the wire, - # not N+1. A forced republish (reconnect / not-yet-READY) is exempt — it - # must reach the broker now. (_end_state_transition leaves the transition - # scope before its own call so that consolidated publish isn't deferred.) + # not N+1. (_end_state_transition leaves the transition scope before its own + # call so that consolidated publish isn't deferred.) if self._transition_depth > 0 and not republish: logger.debug( f"reason=publishDescriptionDeferredInTransition,deviceId={self._id},depth={self._transition_depth}" ) return - # SDK-n83: defensive no-op when the description content (ignoring the - # always-fresh `version` timestamp) is byte-identical to what we last - # published — avoids the redundant ~KB republish and the gratuitous - # INIT→READY flap that forces every subscriber to resync. A forced - # republish is exempt so reconnect always restores the retained topic. - if not republish: - if self._description_content_hash(self.description()) == self._last_description_content_hash: - logger.debug(f"reason=publishDescriptionUnchanged,deviceId={self._id}") - return - - if republish: - self.publish("$description") + try: + description = self.description() + changed = self._description_content_hash(description) != self._last_description_content_hash + except Exception as e: + # Best-effort, like every Device.publish(): one unserializable entry must + # not abort a tree-wide refresh. + logger.exception(f"reason=publishDescriptionComposeFailed,deviceId={self._id},e={e}") + return + if not changed and not republish: + # SDK-n83: nothing to announce; skip the ~KB republish and the INIT flap. + logger.debug(f"reason=publishDescriptionUnchanged,deviceId={self._id}") + return + prior = self._state + if changed and not republish and prior in (DeviceState.READY, DeviceState.SLEEPING): + self.publish_state(DeviceState.INIT) + self.publish("$description", description) + self.publish_state(prior) else: - if self._state == DeviceState.READY: - # Need to transition first to INIT - self.publish_state(DeviceState.INIT) - self.publish("$description") - # Now that we've republished, restore $state to ready - self.publish_state(DeviceState.READY) - else: - # TODO: should we be able to publish if DISCONNECTED, SLEEPING, or LOST? - # If not in READY state, then we don't need to transition to INIT... - logger.info(f"reason=publishDescriptionNotRepublishNotReady,state={self._state.name}") - # Just publish description - self.publish("$description") + self.publish("$description", description) def publish_nodes(self, *, force: bool = True) -> None: # Snapshot — invoked from on_connect() on the MQTT loop thread while @@ -2579,11 +3329,18 @@ def will(self) -> dict: ``declare_lost()`` publishes this same topic and payload explicitly, for a producer that knows it is dying: the will fires only on an unclean disconnect, and the clean disconnect ``stop()`` performs suppresses it. + + ``qos`` and ``retain`` are the tree's: every other ``$state`` publish is + retained at the tree QoS (convention.md:45-47), and a will on the same topic + at a lower QoS would break the ordering a single QoS level guarantees + (convention.md:689-691). """ root = self.root() return { "topic": f"{root.homie_domain()}/{EBUS_HOMIE_VERSION_MAJOR}/{root._id}/$state", "payload": DeviceState.LOST.value, + "qos": root._qos, + "retain": True, } def connect_broker(self) -> None: diff --git a/src/ebus_sdk/transport.py b/src/ebus_sdk/transport.py index 947d6b1..7dbaddf 100644 --- a/src/ebus_sdk/transport.py +++ b/src/ebus_sdk/transport.py @@ -70,6 +70,12 @@ class MqttDeviceTransport(MqttTransport, Protocol): This protocol has a data member (``is_running``), so use ``isinstance`` for runtime checks; ``issubclass`` is unsupported for protocols with non-method members. + + Optional: a ``subscribe`` that takes a keyword-only ``with_retain`` (as + ebus-mqtt-client 0.7.0 does) is called with ``with_retain=True`` for ``/set`` + subscriptions and must then pass the message's retain flag as a third callback + argument, so the device can ignore a retained ``/set``. Without the parameter + the SDK subscribes as before and cannot detect one. """ def is_connected(self) -> bool: ... diff --git a/tests/test_homie_compliance.py b/tests/test_homie_compliance.py new file mode 100644 index 0000000..1ea5b42 --- /dev/null +++ b/tests/test_homie_compliance.py @@ -0,0 +1,909 @@ +"""Homie 5 convention compliance of the device role (issue #95). + +Line numbers in docstrings refer to homieiot/convention@7edc221 convention.md. +""" + +import datetime +import decimal +import enum +import inspect +import json +import logging +import socket +from unittest.mock import MagicMock, patch + +import pytest +from ebus_mqtt_client import MqttClient + +from ebus_sdk.homie import ( + EBUS_HOMIE_DOMAIN, + EBUS_HOMIE_VERSION_MAJOR, + Device, + DeviceState, + Node, + Property, + PropertyDatatype, +) + +BASE = f"{EBUS_HOMIE_DOMAIN}/{EBUS_HOMIE_VERSION_MAJOR}" + + +def _mock_mqtt_client(): + mock = MagicMock() + mock.is_running = True + mock.is_connected.return_value = True + mock.publish.return_value = MagicMock(rc=0) + mock.subscribe.return_value = (0, 1) + return mock + + +def _make_root(device_id="root", **kwargs): + with patch("ebus_sdk.homie.MqttClient.from_config") as from_config: + client = _mock_mqtt_client() + from_config.return_value = client + device = Device(id=device_id, mqtt_cfg={"host": "localhost", "port": 1883}, **kwargs) + return device, client + + +def _publishes(client): + """(topic, payload) for every plain publish, in order.""" + return [(c.args[0], c.args[1]) for c in client.publish.call_args_list] + + +# ── A. Lifecycle and teardown ──────────────────────────────────────────── + + +class TestStopAnnouncesEveryDevice: + """:280-281: every device sends `disconnected` before the clean disconnect.""" + + def test_owned_stop_publishes_disconnected_descendants_first_root_last(self): + root, client = _make_root() + child = Device(id="child", parent=root) + Device(id="grandchild", parent=child) + client.publish.reset_mock() + + root.stop() + + states = [(t, p) for t, p in _publishes(client) if t.endswith("/$state")] + assert states == [ + (f"{BASE}/grandchild/$state", "disconnected"), + (f"{BASE}/child/$state", "disconnected"), + ] + # The root goes last, flushed, before the clean stop. + client.publish_and_flush.assert_called_once() + assert client.publish_and_flush.call_args.args[:2] == (f"{BASE}/root/$state", "disconnected") + client.stop.assert_called_once() + assert root.state() == child.state() == DeviceState.DISCONNECTED + + def test_stop_from_a_child_announces_the_whole_tree(self): + root, client = _make_root() + child = Device(id="child", parent=root) + client.publish.reset_mock() + + child.stop() + + assert (f"{BASE}/child/$state", "disconnected") in _publishes(client) + assert client.publish_and_flush.call_args.args[0] == f"{BASE}/root/$state" + + def test_injected_stop_publishes_disconnected_for_every_device(self): + client = _mock_mqtt_client() + root = Device(id="root", mqttc=client) + Device(id="child", parent=root) + client.reset_mock() + + root.stop() + + assert _publishes(client) == [ + (f"{BASE}/child/$state", "disconnected"), + (f"{BASE}/root/$state", "disconnected"), + ] + client.publish_and_flush.assert_not_called() + client.stop.assert_not_called() + + +class TestSilentStopIsABadDisconnect: + """:281 and :284: a teardown that announced nothing must not disconnect cleanly.""" + + def _recorder(self, client): + order = MagicMock() + order.attach_mock(client.mqttc.loop_stop, "loop_stop") + order.attach_mock(client.mqttc.socket.return_value.shutdown, "shutdown") + order.attach_mock(client.stop, "stop") + return order + + def test_announce_false_drops_the_socket_before_the_client_stop(self): + root, client = _make_root() + order = self._recorder(client) + + root.stop(announce=False) + + names = [c[0] for c in order.mock_calls] + assert names.index("loop_stop") < names.index("shutdown") < names.index("stop") + client.mqttc.socket.return_value.shutdown.assert_called_once_with(socket.SHUT_RDWR) + client.publish_and_flush.assert_not_called() + + def test_declare_lost_then_silent_stop_publishes_lost_and_lets_the_will_fire(self): + root, client = _make_root() + Device(id="child", parent=root) + client.publish.reset_mock() + + root.declare_lost() + root.stop(announce=False) + + assert client.publish_and_flush.call_args.args[:2] == (f"{BASE}/root/$state", "lost") + assert all(p != "disconnected" for _, p in _publishes(client)) + client.mqttc.socket.return_value.shutdown.assert_called_once() + + def test_default_stop_disconnects_cleanly(self): + root, client = _make_root() + root.stop() + client.mqttc.socket.return_value.shutdown.assert_not_called() + client.stop.assert_called_once() + + +class TestWill: + """:45-47 and :689-691: the will carries the tree's QoS and is retained.""" + + def test_will_carries_qos_and_retain(self): + root, _ = _make_root(qos=1) + will = root.will() + assert will["qos"] == 1 + assert will["retain"] is True + + def test_owned_client_is_built_with_the_full_will(self): + with patch("ebus_sdk.homie.MqttClient.from_config") as from_config: + from_config.return_value = _mock_mqtt_client() + root = Device(id="root", mqtt_cfg={"host": "localhost"}) + assert from_config.call_args.kwargs["lwt"]["qos"] == root.qos + + +class TestChildRemovalOrder: + """:635-641: update the parent first, then clear the child starting with `$state`.""" + + def test_parent_reannounces_before_the_child_is_cleared(self): + root, client = _make_root() + child = Device(id="child", parent=root) + child.add_node(child.new_node("n")) + client.publish.reset_mock() + + child.delete() + + events = _publishes(client) + parent_init = events.index((f"{BASE}/root/$state", "init")) + parent_desc = next(i for i, (t, _) in enumerate(events) if t == f"{BASE}/root/$description") + parent_ready = events.index((f"{BASE}/root/$state", "ready")) + child_state = events.index((f"{BASE}/child/$state", "")) + child_desc = events.index((f"{BASE}/child/$description", "")) + assert parent_init < parent_desc < parent_ready < child_state < child_desc + assert "child" not in json.loads(events[parent_desc][1])["children"] + + def test_batched_removal_defers_the_parent_update(self): + root, client = _make_root() + children = [Device(id=f"c{i}", parent=root) for i in range(3)] + client.publish.reset_mock() + + with root.state_transition(): + for child in children: + child.delete() + + root_states = [p for t, p in _publishes(client) if t == f"{BASE}/root/$state"] + assert root_states == ["init", "ready"] + + +class TestDeletedDeviceStaysDeleted: + """:272 and :288: a device without `$state` does not exist; nothing may re-create it.""" + + def test_delete_then_stop_publishes_no_state(self): + root, client = _make_root() + Device(id="child", parent=root) + root.delete() + client.publish.reset_mock() + + root.stop() + + assert not [t for t, _ in _publishes(client) if t.endswith("/$state")] + # The only flushed publish re-clears the root's $state, so delete()'s queued + # clears reach the broker before the disconnect. + client.publish_and_flush.assert_called_once() + assert client.publish_and_flush.call_args.args[:2] == (f"{BASE}/root/$state", "") + client.stop.assert_called_once() + + def test_delete_then_silent_stop_disconnects_cleanly(self): + """A bad disconnect would fire the will and re-create the device as `lost`.""" + root, client = _make_root() + root.delete() + + root.stop(announce=False) + + client.mqttc.socket.return_value.shutdown.assert_not_called() + client.stop.assert_called_once() + + def test_a_reconnect_after_delete_republishes_nothing(self): + root, client = _make_root() + root.add_node(root.new_node("n")) + root.delete() + client.publish.reset_mock() + + root.on_connect() + + assert client.publish.call_args_list == [] + + +# ── B. $description ────────────────────────────────────────────────────── + + +def _descriptions(client, device_id): + return [json.loads(p) for t, p in _publishes(client) if t == f"{BASE}/{device_id}/$description" and p] + + +def _states(client, device_id): + return [p for t, p in _publishes(client) if t == f"{BASE}/{device_id}/$state"] + + +def _invalid_input_warnings(caplog): + return [r for r in caplog.records if "next minor release" in r.getMessage()] + + +class TestDescriptionVersion: + """:215: a new version whenever the document changes, and only then.""" + + def test_version_is_a_52_bit_integer(self): + root, _ = _make_root() + version = root.description()["version"] + assert isinstance(version, int) + assert 0 <= version < 2**52 + + def test_equal_content_has_equal_version(self): + a = Device(id="same") + b = Device(id="same") + assert a.description()["version"] == b.description()["version"] + + def test_a_content_change_changes_the_version(self): + root, _ = _make_root() + before = root.description()["version"] + root.add_node(root.new_node("n")) + assert root.description()["version"] != before + + def test_reconnect_republishes_the_same_document_without_init(self): + root, client = _make_root() + root.add_node(root.new_node("n")) + published = _descriptions(client, "root")[-1] + client.publish.reset_mock() + + root.on_connect() + + assert _descriptions(client, "root") == [published] + assert _states(client, "root") == ["ready"] + + +class TestDescriptionChangesOnlyInPermittedStates: + """:207: `$description` may change only in `init`, `disconnected` or `lost`.""" + + def test_a_change_while_sleeping_is_wrapped_in_init_and_returns_to_sleeping(self): + root, client = _make_root() + root.set_state(DeviceState.SLEEPING) + client.publish.reset_mock() + + root.add_node(root.new_node("n")) + + events = [(t.rsplit("/", 1)[1], p) for t, p in _publishes(client)] + assert events[0] == ("$state", "init") + assert [e for e in events if e[0] == "$description"] + assert events[-1] == ("$state", "sleeping") + assert root.state() == DeviceState.SLEEPING + + def test_a_child_added_to_a_sleeping_parent_wraps_the_parent(self): + root, client = _make_root() + root.set_state(DeviceState.SLEEPING) + client.publish.reset_mock() + + Device(id="child", parent=root) + + assert _states(client, "root") == ["init", "sleeping"] + + def test_publish_description_while_ready_wraps_only_a_change(self): + root, client = _make_root() + client.publish.reset_mock() + root.publish_description() + assert client.publish.call_args_list == [] + + +class TestDescriptionFields: + def test_unset_device_and_node_type_are_omitted(self): + """:218 and :305: `type` is not nullable.""" + device = Device(id="d") + device.add_node(Node(id="n")) + description = device.description() + assert "type" not in description + assert "type" not in description["nodes"]["n"] + + def test_set_types_are_kept(self): + device = Device(id="d", type="t") + device.add_node(Node(id="n", type="nt")) + assert device.description()["type"] == "t" + assert device.description()["nodes"]["n"]["type"] == "nt" + + def test_missing_datatype_warns_and_is_omitted(self, caplog): + """:340: `datatype` is required.""" + with caplog.at_level(logging.WARNING, logger="homie"): + prop = Property(id="p") + assert "datatype" not in prop.description() + assert len(_invalid_input_warnings(caplog)) == 1 + + def test_null_retained_warns_and_is_published_as_false(self, caplog): + """:343: `retained` is a non-null boolean; None always published non-retained.""" + with caplog.at_level(logging.WARNING, logger="homie"): + prop = Property(id="p", datatype=PropertyDatatype.FLOAT, retained=None) + assert prop.description()["retained"] is False + assert prop.retained() is False + assert len(_invalid_input_warnings(caplog)) == 1 + + def test_json_format_dict_is_published_as_a_string(self): + """:395: the JSONschema is a string, not a nested object.""" + schema = {"type": "object"} + prop = Property(id="j", datatype=PropertyDatatype.JSON, format=schema) + assert prop.description()["format"] == json.dumps(schema) + + def test_enum_and_color_without_format_warn(self, caplog): + """:392-393: `enum` and `color` require a format.""" + with caplog.at_level(logging.WARNING, logger="homie"): + Property(id="e", datatype=PropertyDatatype.ENUM) + Property(id="c", datatype=PropertyDatatype.COLOR) + assert len(_invalid_input_warnings(caplog)) == 2 + + def test_enum_format_with_empty_or_duplicate_values_warns(self, caplog): + """:392: at least one value, none empty, no duplicates.""" + with caplog.at_level(logging.WARNING, logger="homie"): + Property(id="e1", datatype=PropertyDatatype.ENUM, format="a,,b") + Property(id="e2", datatype=PropertyDatatype.ENUM, format="a,b,a") + Property(id="ok", datatype=PropertyDatatype.ENUM, format="a,b") + reasons = [r.getMessage() for r in _invalid_input_warnings(caplog)] + assert len(reasons) == 2 + assert "propertyID=e1" in reasons[0] and "propertyID=e2" in reasons[1] + + def test_the_warning_is_once_per_object(self, caplog): + with caplog.at_level(logging.WARNING, logger="homie"): + prop = Property(id="e", datatype=PropertyDatatype.ENUM) + prop.set_format(None) + prop.set_format("") + assert len(_invalid_input_warnings(caplog)) == 1 + + def test_set_format_checks_the_new_format(self, caplog): + prop = Property(id="e", datatype=PropertyDatatype.ENUM, format="a,b") + with caplog.at_level(logging.WARNING, logger="homie"): + prop.set_format("a,a") + assert len(_invalid_input_warnings(caplog)) == 1 + + def test_extras_cannot_add_root_or_parent_to_a_root(self, caplog): + """:220: `root` MUST be omitted on the root device.""" + with caplog.at_level(logging.WARNING, logger="homie"): + device = Device(id="r", description_extras={"root": "x", "parent": "y", "imported-from": "z"}) + description = device.description() + assert "root" not in description and "parent" not in description + assert description["imported-from"] == "z" + assert len(_invalid_input_warnings(caplog)) == 2 + + def test_extras_cannot_override_core_fields(self): + device = Device(id="r", description_extras={"version": 1, "type": "x", "homie": "4.0"}) + description = device.description() + assert description["homie"] == "5.0" + assert description["version"] != 1 + assert "type" not in description + + +class TestNewValuesPrecedeReady: + """:279: `ready` follows every value it vouches for.""" + + def test_add_property_on_a_ready_device(self): + root, client = _make_root() + node = root.new_node("n") + root.add_node(node) + client.publish.reset_mock() + + node.add_property(Property(id="p", value=5, datatype=PropertyDatatype.INTEGER)) + + events = [(t.rsplit("/", 1)[1], p) for t, p in _publishes(client)] + kinds = [e[0] if e[0].startswith("$") else "value" for e in events] + assert kinds == ["$state", "value", "$description", "$state"] + assert events[0][1] == "init" and events[-1][1] == "ready" + + def test_add_node_on_a_ready_device(self): + root, client = _make_root() + node = Node(id="n") + node._properties["p"] = Property(id="p", value=1.5, datatype=PropertyDatatype.FLOAT, node=node) + client.publish.reset_mock() + + root.add_node(node) + + events = [(t.rsplit("/", 1)[1], p) for t, p in _publishes(client)] + kinds = [e[0] if e[0].startswith("$") else "value" for e in events] + assert kinds == ["$state", "value", "$description", "$state"] + + def test_add_property_on_a_detached_node_publishes_only_the_value(self): + """:616: no init/ready flap when the $description does not change.""" + root, client = _make_root() + node = root.new_node("n") + client.publish.reset_mock() + + node.add_property(Property(id="a", value=1, datatype=PropertyDatatype.INTEGER)) + node.add_property(Property(id="b", value=2, datatype=PropertyDatatype.INTEGER)) + + assert _publishes(client) == [(f"{BASE}/root/n/a", "1"), (f"{BASE}/root/n/b", "2")] + + root.add_node(node) + assert _states(client, "root") == ["init", "ready"] + + def test_re_adding_an_identical_property_publishes_only_the_value(self): + root, client = _make_root() + node = root.new_node("n") + root.add_node(node) + node.add_property(Property(id="a", value=1, datatype=PropertyDatatype.INTEGER)) + client.publish.reset_mock() + + node.add_property(Property(id="a", value=1, datatype=PropertyDatatype.INTEGER)) + + assert _publishes(client) == [(f"{BASE}/root/n/a", "1")] + + def test_inside_a_transition_nothing_extra_is_published(self): + root, client = _make_root() + client.publish.reset_mock() + with root.state_transition(): + root.add_node(root.new_node("a")) + root.add_node(root.new_node("b")) + assert _states(client, "root") == ["init", "ready"] + assert len(_descriptions(client, "root")) == 1 + + +class TestDescriptionCacheInvalidation: + def test_clearing_the_description_topic_lets_the_next_publish_through(self): + root, client = _make_root() + root.clear_retained_topic(f"{BASE}/root/$description") + client.publish.reset_mock() + + root.publish_description() + + assert len(_descriptions(client, "root")) == 1 + + def test_refresh_after_invalidation_publishes_ready_once_and_last(self): + """:207 and :279: init first, then the document, values and children, then ready.""" + root, client = _make_root() + node = root.new_node("n") + root.add_node(node) + node.add_property(Property(id="p", value=1.5, datatype=PropertyDatatype.FLOAT)) + Device(id="kid", parent=root) + root.clear_retained_topic(f"{BASE}/root/$description") + client.publish.reset_mock() + + root.refresh_tree() + + events = [(t.split("/", 2)[2], p) for t, p in _publishes(client)] + topics = [t for t, _ in events] + assert _states(client, "root") == ["init", "ready"] + assert events[0] == ("root/$state", "init") and events[-1] == ("root/$state", "ready") + assert topics.index("root/$description") < topics.index("root/n/p") < topics.index("kid/$state") + + +# ── C. Property values ─────────────────────────────────────────────────── + + +def _wired(datatype, **kwargs): + """A property on a node on a READY root with a mock client.""" + root, client = _make_root() + node = root.new_node("n") + root.add_node(node) + prop = Property(id="p", datatype=datatype, **kwargs) + node.add_property(prop) + client.publish.reset_mock() + return prop, client, root + + +def _value_publishes(client): + return [c for c in client.publish.call_args_list if c.args[0] == f"{BASE}/root/n/p"] + + +def _published(datatype, value, **kwargs): + """The payload set_value(value) puts on the wire, or None if it publishes nothing.""" + prop, client, _ = _wired(datatype, **kwargs) + ok = prop.set_value(value) + calls = _value_publishes(client) + assert ok == bool(calls) + return calls[-1].args[1] if calls else None + + +class TestOutboundNumbers: + """:85-101: integer and float payload grammar.""" + + def test_non_finite_floats_are_refused(self): + for value in (float("nan"), float("inf"), float("-inf"), "1e999"): + assert _published(PropertyDatatype.FLOAT, value) is None + + def test_float_exponent_has_no_plus(self): + """:97: only digits, `-`, `e`/`E` and `.`.""" + assert _published(PropertyDatatype.FLOAT, 1e20) == "1e20" + assert _published(PropertyDatatype.FLOAT, 1.5e-7) == "1.5e-07" + assert _published(PropertyDatatype.FLOAT, "1e+5") is None + + def test_floats_and_ints_on_a_float_property(self): + assert _published(PropertyDatatype.FLOAT, 21.5) == "21.5" + assert _published(PropertyDatatype.FLOAT, 5) == "5" + assert _published(PropertyDatatype.FLOAT, "-3.25") == "-3.25" + + def test_non_ascii_digits_are_refused(self): + """:89 and :97: payload digits are 0-9.""" + assert _published(PropertyDatatype.INTEGER, "\u0661\u0662") is None + assert _published(PropertyDatatype.FLOAT, "\u0663") is None + assert _published(PropertyDatatype.DURATION, "PT\u0661S") is None + + def test_decimal_on_a_float_property(self): + assert _published(PropertyDatatype.FLOAT, decimal.Decimal("1.5")) == "1.5" + assert _published(PropertyDatatype.FLOAT, decimal.Decimal("1E+5")) == "1E5" + assert _published(PropertyDatatype.FLOAT, decimal.Decimal("NaN")) is None + assert _published(PropertyDatatype.FLOAT, decimal.Decimal("1E+400")) is None + + def test_float_range_applies_to_integers(self): + """:95-96.""" + assert _published(PropertyDatatype.FLOAT, 10**300) == str(10**300) + assert _published(PropertyDatatype.FLOAT, 10**400) is None + + def test_numpy_float32_keeps_its_rounded_text(self): + np = pytest.importorskip("numpy") + assert _published(PropertyDatatype.FLOAT, np.float32(0.14494), round_to=1) == "0.1" + + def test_booleans_are_refused_on_numeric_properties(self): + assert _published(PropertyDatatype.FLOAT, True) is None + assert _published(PropertyDatatype.INTEGER, False) is None + + def test_integers_must_be_whole_and_64_bit(self): + """:87-90.""" + assert _published(PropertyDatatype.INTEGER, 5.0) == "5" + assert _published(PropertyDatatype.INTEGER, 5.5) is None + assert _published(PropertyDatatype.INTEGER, 2**63 - 1) == str(2**63 - 1) + assert _published(PropertyDatatype.INTEGER, 2**63) is None + assert _published(PropertyDatatype.INTEGER, -(2**63) - 1) is None + assert _published(PropertyDatatype.INTEGER, "-") is None + assert _published(PropertyDatatype.INTEGER, "12") == "12" + + def test_a_refused_value_leaves_the_last_good_one_on_the_broker(self): + prop, client, _ = _wired(PropertyDatatype.FLOAT) + prop.set_value(1.0) + client.publish.reset_mock() + assert prop.set_value(float("nan")) is False + assert client.publish.call_args_list == [] + assert prop.get_last_published_value() == "1.0" + + +class TestOutboundEnumAndEmptyStrings: + def test_enum_values_must_be_in_the_format(self): + """:111.""" + + class Mode(str, enum.Enum): + AUTO = "auto" + + assert _published(PropertyDatatype.ENUM, "auto", format="auto,off") == "auto" + assert _published(PropertyDatatype.ENUM, Mode.AUTO, format="auto,off") == "auto" + assert _published(PropertyDatatype.ENUM, "Auto", format="auto,off") is None + + def test_empty_string_is_0x00_only_for_string(self): + """:65-67: the empty string is valid only for string properties.""" + assert _published(PropertyDatatype.STRING, "") == "\x00" + assert _published(PropertyDatatype.ENUM, "", format="a,b") is None + assert _published(PropertyDatatype.INTEGER, "") is None + assert _published(PropertyDatatype.JSON, "") is None + assert _published(PropertyDatatype.DATETIME, "") is None + + +class TestOutboundEncoders: + def test_datetime_is_iso_8601(self): + """:132.""" + stamp = datetime.datetime(2026, 10, 8, 12, 0, tzinfo=datetime.timezone.utc) + assert _published(PropertyDatatype.DATETIME, stamp) == "2026-10-08T12:00:00+00:00" + assert _published(PropertyDatatype.DATETIME, datetime.date(2026, 10, 8)) == "2026-10-08" + + def test_datetime_strings_are_checked(self): + """:132: calendar, ordinal and week dates, extended or basic, with optional time.""" + for good in ("2026-10-08T12:00:00+00:00", "2026-10-08T12:00Z", "20261008T120000Z", "2026-281", "2026-W41-4"): + assert _published(PropertyDatatype.DATETIME, good) == good, good + for bad in ("garbage", "2026-10-08 12:00", "2026-13-01", "2026-02-30", "2026-10-08T25:00", "\u0662026-10-08"): + assert _published(PropertyDatatype.DATETIME, bad) is None, bad + + def test_duration_is_ptxhxmxs(self): + """:137-144.""" + timedelta = datetime.timedelta + assert _published(PropertyDatatype.DURATION, timedelta(hours=12, minutes=5, seconds=46)) == "PT12H5M46S" + assert _published(PropertyDatatype.DURATION, timedelta(minutes=5)) == "PT5M" + assert _published(PropertyDatatype.DURATION, timedelta(0)) == "PT0S" + assert _published(PropertyDatatype.DURATION, timedelta(days=2)) == "PT48H" + assert _published(PropertyDatatype.DURATION, timedelta(seconds=1.5)) == "PT1.5S" + assert _published(PropertyDatatype.DURATION, timedelta(seconds=-1)) is None + assert _published(PropertyDatatype.DURATION, "PT5M") == "PT5M" + assert _published(PropertyDatatype.DURATION, "PT") is None + assert _published(PropertyDatatype.DURATION, "5 minutes") is None + + def test_color_tuple_uses_the_preferred_format(self): + """:118-123 and :393.""" + assert _published(PropertyDatatype.COLOR, (255, 0, 0), format="rgb,hsv") == "rgb,255,0,0" + assert _published(PropertyDatatype.COLOR, (300, 50, 75.5), format="hsv") == "hsv,300,50,75.5" + assert _published(PropertyDatatype.COLOR, (0.25, 0.34), format="xyz") == "xyz,0.25,0.34" + + def test_color_payloads_are_checked(self): + assert _published(PropertyDatatype.COLOR, "hsv,300,50,75", format="rgb,hsv") == "hsv,300,50,75" + assert _published(PropertyDatatype.COLOR, "rgb,256,0,0", format="rgb") is None + assert _published(PropertyDatatype.COLOR, "rgb, 1,2,3", format="rgb") is None + assert _published(PropertyDatatype.COLOR, "xyz,0.2,0.3", format="rgb") is None + assert _published(PropertyDatatype.COLOR, (1, 2), format="rgb") is None + + def test_json_must_be_an_array_or_object_without_nan(self): + """:150.""" + assert _published(PropertyDatatype.JSON, {"a": 1}) == '{"a": 1}' + assert _published(PropertyDatatype.JSON, [1, 2]) == "[1, 2]" + assert _published(PropertyDatatype.JSON, '{"a":1}') == '{"a":1}' + assert _published(PropertyDatatype.JSON, "hello") is None + assert _published(PropertyDatatype.JSON, 5) is None + assert _published(PropertyDatatype.JSON, '"text"') is None + assert _published(PropertyDatatype.JSON, {"a": float("nan")}) is None + assert _published(PropertyDatatype.JSON, '{"a": NaN}') is None + + +class TestEventProperties: + """:50 and :695: non-retained properties publish at QoS 0 and are never replayed.""" + + def test_an_event_publishes_non_retained_at_qos_0(self): + prop, client, _ = _wired(PropertyDatatype.STRING, retained=False) + prop.set_value("pressed") + (published,) = _value_publishes(client) + assert published.kwargs == {"retain": False, "qos": 0} + + def test_a_reconnect_does_not_replay_the_last_event(self): + prop, client, root = _wired(PropertyDatatype.STRING, retained=False) + prop.set_value("pressed") + client.publish.reset_mock() + root.on_connect() + assert _value_publishes(client) == [] + + def test_adding_a_valued_event_property_publishes_no_event(self): + root, client = _make_root() + node = root.new_node("n") + root.add_node(node) + client.publish.reset_mock() + node.add_property(Property(id="p", value="x", datatype=PropertyDatatype.STRING, retained=False)) + assert _value_publishes(client) == [] + + def test_clearing_an_event_publishes_nothing(self): + prop, client, root = _wired(PropertyDatatype.STRING, retained=False) + prop.set_value("pressed") + client.publish.reset_mock() + prop.set_value(None) + root.delete_node("n") + assert _value_publishes(client) == [] + + +class TestAlerts: + """:512-527 and :289.""" + + def test_publish_alert_uses_the_alert_id_level(self): + root, client = _make_root() + client.publish.reset_mock() + assert root.publish_alert("battery", "Battery is low, at 8%") is True + (published,) = client.publish.call_args_list + assert published.args == (f"{BASE}/root/$alert/battery", "Battery is low, at 8%") + assert published.kwargs["retain"] is True + + def test_clear_alert_deletes_the_topic(self): + root, client = _make_root() + root.publish_alert("battery", "low") + client.publish.reset_mock() + assert root.clear_alert("battery") is True + assert _publishes(client) == [(f"{BASE}/root/$alert/battery", "")] + assert root.alerts() == {} + + def test_delete_clears_raised_alerts(self): + root, client = _make_root() + child = Device(id="child", parent=root) + child.publish_alert("childlost", "gone") + client.publish.reset_mock() + child.delete() + assert (f"{BASE}/child/$alert/childlost", "") in _publishes(client) + + def test_a_reconnect_republishes_raised_alerts(self): + root, client = _make_root() + root.publish_alert("battery", "low") + client.publish.reset_mock() + root.on_connect() + assert (f"{BASE}/root/$alert/battery", "low") in _publishes(client) + + def test_bare_alert_and_bad_input_publish_nothing(self): + root, client = _make_root() + client.publish.reset_mock() + root.publish("$alert", "no id") + assert root.publish_alert("a/b", "x") is False + assert root.publish_alert("", "x") is False + assert root.publish_alert("battery", "") is False + assert client.publish.call_args_list == [] + + +# ── D. /set on the device ──────────────────────────────────────────────── + + +def _settable(datatype, **kwargs): + """A settable property with a recording callback, on a root with a mock client.""" + received = [] + prop, client, root = _wired(datatype, settable=True, set_callback=received.append, **kwargs) + return prop, received + + +def _send(prop, payload, retained=None): + topic = f"{BASE}/root/n/p/set" + data = payload.encode() if isinstance(payload, str) else payload + if retained is None: + prop._settable_callback(topic, data) + else: + prop._settable_callback(topic, data, retained) + + +def _accepts(datatype, payload, **kwargs): + """What the set callback receives for ``payload``, or the string REJECTED.""" + prop, received = _settable(datatype, **kwargs) + _send(prop, payload) + return received[0] if received else "REJECTED" + + +class TestInboundValidation: + """Invalid /set payloads are dropped before the callback.""" + + def test_integer_grammar_and_range(self): + assert _accepts(PropertyDatatype.INTEGER, "42") == "42" + assert _accepts(PropertyDatatype.INTEGER, "-7") == "-7" + for bad in ("1.5", "abc", "-", "", " 1", "\x00", str(2**63)): + assert _accepts(PropertyDatatype.INTEGER, bad) == "REJECTED", bad + + def test_float_grammar(self): + assert _accepts(PropertyDatatype.FLOAT, "21.5") == "21.5" + assert _accepts(PropertyDatatype.FLOAT, "1e5") == "1e5" + for bad in ("NaN", "Infinity", "1e+5", "1,5", "", "1e999"): + assert _accepts(PropertyDatatype.FLOAT, bad) == "REJECTED", bad + + def test_numbers_round_to_the_step_then_check_min_max(self): + """:397-413, with floor(x + 0.5) rounding.""" + assert _accepts(PropertyDatatype.INTEGER, "5", format="0:10:2") == "6" + assert _accepts(PropertyDatatype.INTEGER, "5", format=":10:2") == "6" + assert _accepts(PropertyDatatype.INTEGER, "4", format="0:10:2") == "4" + # 11 rounds to 12, which is outside the range: checked after rounding. + assert _accepts(PropertyDatatype.INTEGER, "11", format="0:10:2") == "REJECTED" + assert _accepts(PropertyDatatype.FLOAT, "0.25", format="0:1:0.1") == "0.3" + assert _accepts(PropertyDatatype.FLOAT, "101", format="0:100") == "REJECTED" + assert _accepts(PropertyDatatype.FLOAT, "-0.5", format="0:") == "REJECTED" + + def test_the_current_value_is_the_step_base_without_min_or_max(self): + assert _accepts(PropertyDatatype.INTEGER, "6", format="::4", value=1) == "5" + + def test_non_ascii_digits_are_rejected(self): + assert _accepts(PropertyDatatype.INTEGER, "\u0661") == "REJECTED" + assert _accepts(PropertyDatatype.FLOAT, "\u0663.5") == "REJECTED" + assert _accepts(PropertyDatatype.DURATION, "PT\u0661S") == "REJECTED" + + def test_datetime_must_be_iso_8601(self): + assert _accepts(PropertyDatatype.DATETIME, "2026-10-08T12:00:00Z") == "2026-10-08T12:00:00Z" + for bad in ("banana", "2026-10-08 12:00", ""): + assert _accepts(PropertyDatatype.DATETIME, bad) == "REJECTED", bad + + def test_boolean_is_case_sensitive(self): + assert _accepts(PropertyDatatype.BOOLEAN, "true") == "true" + for bad in ("TRUE", "1", "yes", ""): + assert _accepts(PropertyDatatype.BOOLEAN, bad) == "REJECTED", bad + + def test_enum_membership(self): + assert _accepts(PropertyDatatype.ENUM, "auto", format="auto,off") == "auto" + assert _accepts(PropertyDatatype.ENUM, "Auto", format="auto,off") == "REJECTED" + + def test_json_must_be_an_array_or_object(self): + assert _accepts(PropertyDatatype.JSON, "[1, 2]") == [1, 2] + for bad in ("5", '"s"', "null", "{", '{"a": NaN}'): + assert _accepts(PropertyDatatype.JSON, bad) == "REJECTED", bad + + def test_empty_string_decoding_is_for_string_only(self): + assert _accepts(PropertyDatatype.STRING, "\x00") == "" + assert _accepts(PropertyDatatype.ENUM, "\x00", format="a,b") == "REJECTED" + + def test_a_rejection_is_a_warning(self, caplog): + prop, _ = _settable(PropertyDatatype.INTEGER) + with caplog.at_level(logging.WARNING, logger="homie"): + _send(prop, "1.5") + assert any("propertySetRejected" in r.getMessage() for r in caplog.records) + + +class TestRetainedSetIsIgnored: + """:48 and :483: a retained /set is a stale command replayed at subscribe.""" + + def test_a_retained_set_is_ignored_with_a_warning(self, caplog): + prop, received = _settable(PropertyDatatype.INTEGER) + with caplog.at_level(logging.WARNING, logger="homie"): + _send(prop, "5", retained=True) + assert received == [] + assert any("propertySetRetainedIgnored" in r.getMessage() for r in caplog.records) + + def test_a_live_set_runs(self): + prop, received = _settable(PropertyDatatype.INTEGER) + _send(prop, "5", retained=False) + assert received == ["5"] + + def test_subscribes_with_retain_when_the_transport_supports_it(self): + class RetainAwareClient: + is_running = True + + def __init__(self): + self.subscriptions = [] + + def is_connected(self): + return True + + def publish(self, topic, data, qos=1, retain=False): + return None + + def subscribe(self, sub, param, qos=1, *, with_retain=False): + self.subscriptions.append((sub, with_retain)) + + client = RetainAwareClient() + root = Device(id="root", mqttc=client) + node = root.new_node("n") + root.add_node(node) + node.add_property(Property(id="p", datatype=PropertyDatatype.INTEGER, settable=True, set_callback=print)) + assert client.subscriptions == [(f"{BASE}/root/n/p/set", True)] + + def test_an_older_transport_subscribes_without_it(self, caplog): + class OlderClient: + is_running = True + + def __init__(self): + self.subscriptions = [] + + def is_connected(self): + return True + + def publish(self, topic, data, qos=1, retain=False): + return None + + def subscribe(self, sub, param, qos=1): + self.subscriptions.append(sub) + + client = OlderClient() + root = Device(id="root", mqttc=client) + node = root.new_node("n") + root.add_node(node) + with caplog.at_level(logging.DEBUG, logger="homie"): + node.add_property(Property(id="p", datatype=PropertyDatatype.INTEGER, settable=True, set_callback=print)) + assert client.subscriptions == [f"{BASE}/root/n/p/set"] + assert any("propertySetSubscribeNoRetainFlag" in r.getMessage() for r in caplog.records) + + def test_end_to_end_through_ebus_mqtt_client(self, mock_paho): + """The installed ebus-mqtt-client, whichever it is, delivers or omits the flag consistently.""" + received = [] + root = Device(id="root", mqtt_cfg={"host": "localhost", "port": 1883}) + root.start_mqtt_client() + node = root.new_node("n") + root.add_node(node) + node.add_property( + Property(id="p", datatype=PropertyDatatype.INTEGER, settable=True, set_callback=received.append) + ) + supported = "with_retain" in inspect.signature(MqttClient.subscribe).parameters + on_message = mock_paho.on_message + for retain in (True, False): + msg = MagicMock(topic=f"{BASE}/root/n/p/set", payload=b"5", retain=retain) + on_message(mock_paho, None, msg) + assert received == (["5"] if supported else ["5", "5"]) + root.stop() + + +class TestSupportsTargetIsInert: + """:466 and :470: the `$target` stub must not run before the acceptance decision.""" + + def test_one_warning_at_construction(self, caplog): + with caplog.at_level(logging.WARNING, logger="homie"): + Property(id="p", datatype=PropertyDatatype.INTEGER, supports_target=True) + assert sum("propertySupportsTargetInert" in r.getMessage() for r in caplog.records) == 1 + + def test_set_does_not_call_the_target_stub(self): + prop, received = _settable(PropertyDatatype.INTEGER, supports_target=True) + with patch.object(Property, "publish_target_value") as stub: + _send(prop, "5") + stub.assert_not_called() + assert received == ["5"]