From d4f0d0c710e80554ac197d5b939dcfe869ef9ac4 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <209825114+claude[bot]@users.noreply.github.com> Date: Fri, 2 Oct 2026 11:02:55 +0300 Subject: [PATCH] feat(backend): LED strips are stored, normalised and transferred (#780) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 1 of #780 — the data model. A space carries an optional `led_strips: [{id, points, marker, active?}]` (custom_components/houseplan/ led_strips.py, pure, strict mypy). - Type schema inside SPACE_SCHEMA: 2–50 finite numeric points (no strings, booleans, NaN or off-canvas values), ≤50 strips per space including hidden shapes, strict boolean `active`, marker = non-empty string or null. - A config-level step after coordinate canonicalisation judges the shape (two distinct points, non-zero length, a closed strip needs three distinct vertices, a hidden shape needs a marker, unique ids per space) and the links in the order the spec fixes: duplicates are rejected before any normalisation (two links to the same missing id still conflict); a link to a marker that is not live becomes an unbound strip (marker null, active true, id and points kept), so a client that does not know strips can delete a bound marker without its save failing; a live marker with an empty space adopts the strip's space; a non-empty foreign space rejects the write. - config/set answers with `led_strips: {unbound, space_adopted}` when the write was normalised, so a new client re-reads; old clients ignore it. - Space import remaps links through the marker id map; a skipped or virtualised duplicate leaves the strip unbound; a coinciding old id never binds. Plan-only export keeps geometry and nulls every link. Import details report `unbound_led_strips`, computed by the server, never read from the file. - Coordinates get JSON-noise cleanup only, like stairs (face contacts are off-lattice), in both canonicalisers with a shared fixture case. - Support package: counters only (total/unbound/hidden), no coordinates or ids. Tests: tests_backend/test_led_strips.py (41, pure), test_ha_import_export (6 cases: full round trip with a hidden shape, orphan count, remap against a coinciding id, skip and virtual duplicates, plan-only), test_ha_websocket (old client deletes a bound marker → save stands, counters, foreign space rejects without a new revision). Full backend with the HA harness: 969 passed. Mutating the shape check, the duplicate check, the orphan normalisation or the space adoption each turns the pure suite red. Issue: #780 User-Visible: no --- .../houseplan/coordinate_canonicalization.py | 16 + custom_components/houseplan/import_export.py | 32 ++ custom_components/houseplan/led_strips.py | 287 ++++++++++++++++++ .../houseplan/support_package.py | 12 + custom_components/houseplan/validation.py | 9 + custom_components/houseplan/websocket_api.py | 11 +- pyproject.toml | 1 + scripts/config-schema.json | 27 ++ src/coordinate-canonicalization.ts | 14 + .../fixtures/coordinate-canonicalization.json | 14 + tests_backend/test_ha_import_export.py | 117 +++++++ tests_backend/test_ha_websocket.py | 57 ++++ tests_backend/test_led_strips.py | 279 +++++++++++++++++ 13 files changed, 875 insertions(+), 1 deletion(-) create mode 100755 custom_components/houseplan/led_strips.py create mode 100755 tests_backend/test_led_strips.py diff --git a/custom_components/houseplan/coordinate_canonicalization.py b/custom_components/houseplan/coordinate_canonicalization.py index d44299f6..40dae0c9 100644 --- a/custom_components/houseplan/coordinate_canonicalization.py +++ b/custom_components/houseplan/coordinate_canonicalization.py @@ -86,6 +86,16 @@ def _lattice_points(value: Any) -> None: _lattice_point(point) +def _scalar_points(value: Any) -> None: + if not isinstance(value, list): + return + for point in value: + if not isinstance(point, list): + continue + for index in range(min(2, len(point))): + point[index] = canonicalize_number(point[index]) + + def canonicalize_position(position: Any) -> Any: """Canonicalise lattice x/y in one layout record, preserving metadata.""" result = copy.deepcopy(position) @@ -182,6 +192,12 @@ def canonicalize_config_geometry(config: Any) -> Any: _lattice_point(span.get("a")) _lattice_point(span.get("b")) + for strip in _records(space.get("led_strips")): + # #780: LED strips snap to physical wall faces, like stairs and + # furniture: remove JSON noise only, never pull a face contact + # onto a lattice node. + _scalar_points(strip.get("points")) + for marker in _records(root.get("markers")): _scalar_fields(marker, ("angle",)) diff --git a/custom_components/houseplan/import_export.py b/custom_components/houseplan/import_export.py index 5c18edfe..d7da7776 100644 --- a/custom_components/houseplan/import_export.py +++ b/custom_components/houseplan/import_export.py @@ -35,6 +35,7 @@ from .const import ( PLANS_URL, VERSION, ) +from .led_strips import led_strip_link_report, plan_only_strips, unbind_strips from .radar_validation import validate_marker_radars from .store import HouseplanData from .validation import ( @@ -294,6 +295,10 @@ def _project_plan_only_space(space: dict[str, Any]) -> dict[str, Any]: projected["decor"] = [ _project_plan_only_decor(shape) for shape in space.get("decor") or [] ] + led_strips = plan_only_strips(space) + if led_strips is not None: + # #780: plan-only keeps the strip geometry and never a device link. + projected["led_strips"] = led_strips if "stairs" in space: projected["stairs"] = [ _pick_fields(stair, ( @@ -694,6 +699,12 @@ def parse_document(raw: bytes) -> dict[str, Any]: # before an import token can be issued. if model > 0: config_candidate["model_version"] = model + # #780: the schema step unbinds strips whose marker is not in the + # document; count them before it does, for the import summary. + unbound_led_strips = ( + led_strip_link_report(config_candidate)["unbound"] + if isinstance(config_candidate, dict) else 0 + ) config = CONFIG_SCHEMA(config_candidate) config.pop("model_version", None) except (vol.Invalid, TypeError, ValueError) as err: @@ -730,6 +741,10 @@ def parse_document(raw: bytes) -> dict[str, Any]: **(document.get("transfer") or {}), "dropped_marker_links": dropped_marker_links, } + # Computed here, never trusted from the file. + document["transfer"].pop("unbound_led_strips", None) + if unbound_led_strips: + document["transfer"]["unbound_led_strips"] = unbound_led_strips if plan_only: _validate_plan_only_document(document, config, layout, placement) if len(json.dumps( @@ -1608,6 +1623,14 @@ def build_space_merge( marker.pop("value_source", None) dropped_marker_links += 1 + # #780: an LED strip follows its marker through the same id remap; a + # marker that was skipped or virtualised by the duplicate policy is not + # transferred, so the strip arrives unbound (geometry kept, no device). + unbound_led_strips = unbind_strips(space, remap={ + old_id: new_id for old_id, new_id in marker_map.items() + if old_id not in virtualized_targets + }) + output_markers_by_id = { str(marker.get("id")): marker for marker in output_markers if marker.get("id") is not None @@ -1709,6 +1732,7 @@ def build_space_merge( "virtualized": virtualized, "orphan_markers": len(output_markers) - len(marker_map), "dropped_marker_links": dropped_marker_links, + "unbound_led_strips": unbound_led_strips, "repaired_target_refs": repaired_target_refs, "preserved_unresolved_refs": sum( int(value) for value in reference_report["preservedUnresolved"].values() @@ -1914,6 +1938,14 @@ def _materialize_import_candidate( details = { "dropped_marker_links": _transfer_dropped_marker_links(prepared), } + # #780: strips unbound by the document itself (counted while parsing) + # plus those the transfer unbound (space merge) or the schema step will. + transfer = prepared.get("transfer") if isinstance(prepared.get("transfer"), dict) else {} + details["unbound_led_strips"] = ( + int(details.get("unbound_led_strips", 0)) + + int(transfer.get("unbound_led_strips", 0) or 0) + + led_strip_link_report(config)["unbound"] + ) try: config = CONFIG_SCHEMA(config) except vol.Invalid as err: diff --git a/custom_components/houseplan/led_strips.py b/custom_components/houseplan/led_strips.py new file mode 100755 index 00000000..19b0546c --- /dev/null +++ b/custom_components/houseplan/led_strips.py @@ -0,0 +1,287 @@ +"""LED strips: stored shape, link normalisation and transfer rules (#780). + +A strip is optional geometry of a space:: + + space.led_strips = [{"id": str, "points": [[x, y], ...], + "marker": str | None, "active": bool (optional)}] + +``active`` is the representation of the bound marker: absent/``True`` draws +the strip, ``False`` keeps the shape hidden while the same marker is shown as +an ordinary icon. It is neither ``marker.hidden`` nor the power state. + +The module is pure (voluptuous only), so the schema, the write-path +normalisation and the import/plan-only projections are exercised by the pure +pytest suite without Home Assistant. + +Write-path order (ТЗ #780 §9): + +1. structure, types and limits (``LED_STRIPS_SCHEMA`` inside ``SPACE_SCHEMA``); +2. per-space invariants and duplicate links — **before** any normalisation, so + two strips pointing at the same missing id are still a conflict; +3. normalisation: a link to a marker that is not live in the resulting config + becomes an unbound strip (``marker: None, active: True``) with its id and + points intact — a client that does not know about strips may delete a bound + marker and its save must not fail; a live marker with an empty ``space`` + adopts the strip's space; +4. referential check: a live marker with a non-empty foreign ``space`` + rejects the whole write. +""" +from __future__ import annotations + +import math +from collections.abc import Iterator +from typing import Any + +import voluptuous as vol + +MAX_LED_STRIPS = 50 +MAX_LED_POINTS = 50 +LED_ID_MAX = 64 +LED_MARKER_MAX = 500 +CANVAS_LIMIT = 5000.0 + +LED_KEY = "led_strips" + + +def _coordinate(value: Any) -> float: + """A stored coordinate: a real finite number, never a string or a bool.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise vol.Invalid("LED strip coordinate must be a number") + number = float(value) + if not math.isfinite(number): + raise vol.Invalid("LED strip coordinate must be finite") + if abs(number) > CANVAS_LIMIT: + raise vol.Invalid("LED strip coordinate is outside the canvas") + return number + + +def _strict_bool(value: Any) -> bool: + if not isinstance(value, bool): + raise vol.Invalid("LED strip active must be a boolean") + return value + + +def _marker_ref(value: Any) -> str | None: + if value is None: + return None + if not isinstance(value, str) or not value or len(value) > LED_MARKER_MAX: + raise vol.Invalid("LED strip marker must be a non-empty string or null") + return value + + +def _strip_id(value: Any) -> str: + if not isinstance(value, str) or not value or len(value) > LED_ID_MAX: + raise vol.Invalid("LED strip id must be a non-empty string") + return value + + +_POINT = vol.All([_coordinate], vol.Length(min=2, max=2)) + +LED_STRIP_SCHEMA = vol.Schema( + { + vol.Required("id"): _strip_id, + vol.Required("points"): vol.All( + list, vol.Length(min=2, max=MAX_LED_POINTS), [_POINT], + ), + vol.Optional("marker", default=None): _marker_ref, + vol.Optional("active"): _strict_bool, + }, + extra=vol.ALLOW_EXTRA, +) + +LED_STRIPS_SCHEMA = vol.All(list, vol.Length(max=MAX_LED_STRIPS), [LED_STRIP_SCHEMA]) + + +def _same(a: list[Any], b: list[Any]) -> bool: + return float(a[0]) == float(b[0]) and float(a[1]) == float(b[1]) + + +def strip_is_closed(points: list[Any]) -> bool: + return len(points) >= 4 and _same(points[0], points[-1]) + + +def polyline_length(points: list[Any]) -> float: + total = 0.0 + for index in range(1, len(points)): + ax, ay = points[index - 1] + bx, by = points[index] + total += math.hypot(float(bx) - float(ax), float(by) - float(ay)) + return total + + +def _distinct_vertices(points: list[Any]) -> int: + seen: set[tuple[float, float]] = set() + for point in points: + seen.add((float(point[0]), float(point[1]))) + return len(seen) + + +def validate_strip_shape(strip: dict[str, Any]) -> None: + """Geometry invariants that the type schema cannot express.""" + points = strip["points"] + if _distinct_vertices(points) < 2 or not polyline_length(points) > 0: + raise vol.Invalid("LED strip needs at least two distinct points") + if _same(points[0], points[-1]) and len(points) > 2: + # A repeated first point closes the strip: three distinct vertices. + if _distinct_vertices(points[:-1]) < 3: + raise vol.Invalid("a closed LED strip needs three distinct vertices") + if strip.get("active") is False and not isinstance(strip.get("marker"), str): + raise vol.Invalid("a hidden LED strip shape must belong to a marker") + + +class LedStripLinkError(ValueError): + """A write that cannot be normalised: duplicate or foreign link.""" + + code = "invalid_led_strip" + + def __init__(self, reason: str) -> None: + # Marker ids may carry user-controlled HA identifiers: a stable + # message without the value, like DuplicateMarkerIdError. + super().__init__(reason) + self.reason = reason + + +def _live_markers(config: dict[str, Any]) -> dict[str, dict[str, Any]]: + live: dict[str, dict[str, Any]] = {} + for marker in config.get("markers") or []: + if not isinstance(marker, dict) or marker.get("removed") is True: + continue + marker_id = marker.get("id") + if isinstance(marker_id, str) and marker_id and marker_id not in live: + live[marker_id] = marker + return live + + +def _strips(config: dict[str, Any]) -> Iterator[tuple[dict[str, Any], dict[str, Any]]]: + for space in config.get("spaces") or []: + if not isinstance(space, dict): + continue + for strip in space.get(LED_KEY) or []: + if isinstance(strip, dict): + yield space, strip + + +def led_strip_link_report(config: dict[str, Any]) -> dict[str, int]: + """What the normaliser would change, without changing anything. + + The write path answers the client with these counters so a new client + knows to re-read; an old client ignores the extra keys. + """ + live = _live_markers(config) + unbound = adopted = 0 + for _space, strip in _strips(config): + ref = strip.get("marker") + if not isinstance(ref, str) or not ref: + continue + marker = live.get(ref) + if marker is None: + unbound += 1 + elif marker.get("space") in (None, ""): + adopted += 1 + return {"unbound": unbound, "space_adopted": adopted} + + +def normalize_led_strip_links(config: dict[str, Any]) -> dict[str, Any]: + """Validate and normalise every strip of a whole configuration in place. + + Runs after the type schema and the coordinate canonicalisation, so the + shape invariants judge the stored numbers. + """ + taken: set[str] = set() + for space in config.get("spaces") or []: + if not isinstance(space, dict): + continue + ids: set[str] = set() + for strip in space.get(LED_KEY) or []: + if strip["id"] in ids: + raise vol.Invalid("LED strip ids must be unique within a space") + ids.add(strip["id"]) + validate_strip_shape(strip) + ref = strip.get("marker") + if isinstance(ref, str): + if ref in taken: + raise LedStripLinkError("a marker is bound to more than one LED strip") + taken.add(ref) + live = _live_markers(config) + for space, strip in _strips(config): + ref = strip.get("marker") + if not isinstance(ref, str): + strip["marker"] = None + continue + marker = live.get(ref) + if marker is None: + strip["marker"] = None + strip["active"] = True + continue + owner = marker.get("space") + if owner in (None, ""): + marker["space"] = space["id"] + elif owner != space["id"]: + raise LedStripLinkError("an LED strip is bound to a marker of another space") + return config + + +def config_led_strip_links(config: dict[str, Any]) -> dict[str, Any]: + """voluptuous step for CONFIG_SCHEMA: ``LedStripLinkError`` → ``vol.Invalid``.""" + try: + return normalize_led_strip_links(config) + except LedStripLinkError as err: + raise vol.Invalid(f"{LedStripLinkError.code}: {err.reason}") from err + + +def unbind_strips(space: dict[str, Any], keep: set[str] | None = None, + remap: dict[str, str] | None = None) -> int: + """Transfer rule for imports/copies: remap a link or make the strip unbound. + + ``remap`` maps the source marker id to the id it received in the target; + a marker that was not transferred (skipped duplicate, plan-only, dropped) + leaves an unbound strip with ``active: True``. Returns how many strips + became unbound, for the import summary. + """ + unbound = 0 + for strip in space.get(LED_KEY) or []: + if not isinstance(strip, dict): + continue + ref = strip.get("marker") + if not isinstance(ref, str) or not ref: + strip["marker"] = None + continue + target = (remap or {}).get(ref) + if target is None and keep is not None and ref in keep: + target = ref + if target is None: + strip["marker"] = None + strip["active"] = True + unbound += 1 + else: + strip["marker"] = target + return unbound + + +def plan_only_strips(space: dict[str, Any]) -> list[dict[str, Any]] | None: + """Plan-only export: geometry stays, device links never leak.""" + if LED_KEY not in space: + return None + projected = [] + for strip in space.get(LED_KEY) or []: + if not isinstance(strip, dict): + continue + projected.append({ + "id": strip.get("id"), + "points": [list(point) for point in strip.get("points") or []], + "marker": None, + "active": True, + }) + return projected + + +def led_strip_counts(config: dict[str, Any]) -> dict[str, int]: + """Support-package counters: no coordinates, names or HA identifiers.""" + total = unbound = hidden = 0 + for _space, strip in _strips(config): + total += 1 + if not isinstance(strip.get("marker"), str): + unbound += 1 + elif strip.get("active") is False: + hidden += 1 + return {"led_strips": total, "led_strips_unbound": unbound, "led_strips_hidden": hidden} diff --git a/custom_components/houseplan/support_package.py b/custom_components/houseplan/support_package.py index 094f027f..be5993da 100644 --- a/custom_components/houseplan/support_package.py +++ b/custom_components/houseplan/support_package.py @@ -22,6 +22,7 @@ from .const import ( MAX_SUPPORT_ATTACHMENT_BYTES, PLAN_MODEL_VERSION, ) +from .led_strips import led_strip_counts PACKAGE_FORMAT = "houseplan-support-package" PACKAGE_VERSION = 1 @@ -457,10 +458,21 @@ def _summary(config: object, layout: object) -> dict[str, Any]: "lifecycle": dict(sorted(lifecycles.items())), "binding": dict(sorted(bindings.items())), }, + # #780: counters only — no coordinates, room names or HA identifiers. + "led_strips": _led_strip_summary(spaces), "layout_entries": len(layout) if isinstance(layout, dict) else 0, } +def _led_strip_summary(spaces: list[Any]) -> dict[str, int]: + counts = led_strip_counts({"spaces": [space for space in spaces if isinstance(space, dict)]}) + return { + "total": counts["led_strips"], + "unbound": counts["led_strips_unbound"], + "hidden": counts["led_strips_hidden"], + } + + def _binding_kind(value: object) -> str: text = str(value or "") if text == "virtual": diff --git a/custom_components/houseplan/validation.py b/custom_components/houseplan/validation.py index 3356f3e0..af8a6e7c 100644 --- a/custom_components/houseplan/validation.py +++ b/custom_components/houseplan/validation.py @@ -17,6 +17,10 @@ from custom_components.houseplan.coordinate_canonicalization import ( canonicalize_layout_geometry, canonicalize_position, ) +from custom_components.houseplan.led_strips import ( + LED_STRIPS_SCHEMA, + config_led_strip_links, +) from custom_components.houseplan.vacuum_routes import validate_marker_routes # ---------- limits and extension sets ---------- @@ -1877,6 +1881,8 @@ SPACE_SCHEMA = vol.All(vol.Schema( # (HP-1454-05): relying on a modern client to strip an unbounded legacy # list is not a limit, it is a hope. `Remove` returns the key stripped. vol.Remove("segments"): object, + # #780: optional LED strip shapes of the space (led_strips.py). + vol.Optional("led_strips"): LED_STRIPS_SCHEMA, }, extra=vol.ALLOW_EXTRA, ), _space_geometry_invariants) @@ -2353,4 +2359,7 @@ CONFIG_SCHEMA = vol.All( ), canonicalize_config_geometry, _config_wall_segment_invariants, + # #780: after canonicalisation — shape invariants judge stored numbers; + # orphan links become unbound strips, foreign/duplicate links reject. + config_led_strip_links, ) diff --git a/custom_components/houseplan/websocket_api.py b/custom_components/houseplan/websocket_api.py index 31908928..f294ceed 100755 --- a/custom_components/houseplan/websocket_api.py +++ b/custom_components/houseplan/websocket_api.py @@ -70,6 +70,7 @@ from .import_export import ( revalidate_candidate, ) from .junction_limits import JunctionLimitError, validate_junction_limits +from .led_strips import led_strip_link_report from .plans import ( QuotaError, collect_attachments, @@ -1681,6 +1682,11 @@ async def ws_config_set(hass: HomeAssistant, connection, msg: dict[str, Any]) -> radar_registry = radar_registry_evidence(hass) + # #780: what the write path will normalise in LED strip links (an + # orphan becomes unbound, an empty marker.space adopts the strip's + # space). Reported back so a new client re-reads; old ones ignore it. + led_report = led_strip_link_report(msg["config"]) + def _validate_config_cpu(): def _normalize(candidate): validate_wall_model_transition(candidate, data.get("config")) @@ -1789,7 +1795,10 @@ async def ws_config_set(hass: HomeAssistant, connection, msg: dict[str, Any]) -> from .repairs import async_check_plan_files hass.async_create_task(async_check_plan_files(hass, entry)) - connection.send_result(msg["id"], {"ok": True, "rev": new_rev}) + result: dict[str, Any] = {"ok": True, "rev": new_rev} + if any(led_report.values()): + result["led_strips"] = led_report + connection.send_result(msg["id"], result) # ---------------- whole-plan maintenance ---------------- diff --git a/pyproject.toml b/pyproject.toml index 22b075a5..859e3635 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,5 +42,6 @@ module = [ "custom_components.houseplan.frontend_asset_manifest", "custom_components.houseplan.junction_limits", "custom_components.houseplan.plans", + "custom_components.houseplan.led_strips", ] strict = true diff --git a/scripts/config-schema.json b/scripts/config-schema.json index 75d3eb60..38ff612c 100644 --- a/scripts/config-schema.json +++ b/scripts/config-schema.json @@ -888,6 +888,33 @@ "required": true, "type": "str" }, + "config.spaces[].led_strips": { + "required": false + }, + "config.spaces[].led_strips[]": { + "allowExtra": true + }, + "config.spaces[].led_strips[].active": { + "opaque": "", + "required": false + }, + "config.spaces[].led_strips[].id": { + "opaque": "", + "required": true + }, + "config.spaces[].led_strips[].marker": { + "default": { + "factory": "NoneType" + }, + "opaque": "", + "required": false + }, + "config.spaces[].led_strips[].points": { + "required": true + }, + "config.spaces[].led_strips[].points[][]": { + "opaque": "" + }, "config.spaces[].open_spans": { "required": false }, diff --git a/src/coordinate-canonicalization.ts b/src/coordinate-canonicalization.ts index 5c34b086..69da6c78 100644 --- a/src/coordinate-canonicalization.ts +++ b/src/coordinate-canonicalization.ts @@ -96,6 +96,16 @@ function isDecorBoxKind(value: unknown): value is DecorBoxKind { && (DECOR_BOX_KINDS as readonly string[]).includes(value); } +function scalarPoints(value: unknown): void { + if (!Array.isArray(value)) return; + for (const point of value) { + if (!Array.isArray(point)) continue; + for (let index = 0; index < Math.min(2, point.length); index++) { + point[index] = canonicalizeNumber(point[index]); + } + } +} + function scalarFields(item: JsonRecord, names: readonly string[]): void { for (const name of names) { if (Object.prototype.hasOwnProperty.call(item, name)) { @@ -382,6 +392,10 @@ export function canonicalizeConfigGeometryInPlace(config: T): T { latticePoint(span.a); latticePoint(span.b); } + + // #780: LED strips snap to physical wall faces like stairs: JSON noise + // only, a face contact never moves onto a lattice node. + for (const strip of records(space.led_strips)) scalarPoints(strip.points); } for (const marker of records(root.markers)) scalarFields(marker, ['angle']); diff --git a/test/fixtures/coordinate-canonicalization.json b/test/fixtures/coordinate-canonicalization.json index 120ba80f..88d717a3 100644 --- a/test/fixtures/coordinate-canonicalization.json +++ b/test/fixtures/coordinate-canonicalization.json @@ -171,6 +171,13 @@ "b": [0.6000000006, 0.3000000004] } ], + "led_strips": [ + { + "id": "led-1", + "points": [[0.3000000004, 0.4166666671], [0.1234567896, -0.0000000004]], + "marker": null + } + ], "future": { "numeric": 0.1234567896 } @@ -386,6 +393,13 @@ "b": [0.6, 0.3] } ], + "led_strips": [ + { + "id": "led-1", + "points": [[0.3, 0.416666667], [0.12345679, 0]], + "marker": null + } + ], "future": { "numeric": 0.1234567896 } diff --git a/tests_backend/test_ha_import_export.py b/tests_backend/test_ha_import_export.py index 596e1f8c..2f2cfeee 100644 --- a/tests_backend/test_ha_import_export.py +++ b/tests_backend/test_ha_import_export.py @@ -3032,3 +3032,120 @@ def test_issue_162_space_export_keeps_legacy_calibration_untouched(tmp_path: Pat vacuum = document["payload"]["config"]["markers"][0]["vacuum"] assert vacuum == {"source": "camera.robot", "calibration": {"m1": [1, 0, 0, 0, 1, 0]}} assert document["transfer"]["dropped_marker_links"] == 0 + + +# ---------- #780: LED strips through export/import ---------- + +def _led_config() -> dict: + config = _config() + config["spaces"][0]["led_strips"] = [ + {"id": "led-a", "points": [[0.1, 0.1], [0.5, 0.1], [0.5, 0.4]], "marker": "lamp"}, + {"id": "led-b", "points": [[0.2, 0.6], [0.8, 0.6]], "marker": None}, + ] + return config + + +def _led_document(tmp_path: Path, kind: str, *, plan_only: bool = False, config=None) -> dict: + document, _ = create_export( + SimpleNamespace(instance_id="instance-a"), + {"config": config or _led_config(), "rev": 2}, + {"layout": {"lamp": {"x": 0.4, "y": 0.5, "s": "ground"}}, "rev": 3}, + kind=kind, space_id="ground" if kind == "space" else None, + plan_only=plan_only, card_version="review", config_root=tmp_path, + ) + return parse_document(json.dumps(document).encode()) + + +def test_issue_780_full_round_trip_keeps_geometry_links_and_hidden_shapes(tmp_path: Path) -> None: + config = _led_config() + config["spaces"][0]["led_strips"][0]["active"] = False + document = _led_document(tmp_path, "full", config=config) + runtime = SimpleNamespace(instance_id="instance-a", import_previews={}) + response = create_preview( + runtime, json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip", + current_config_data={"config": {"spaces": [], "markers": []}, "rev": 0}, + current_layout_data={"layout": {}, "rev": 0}, config_root=tmp_path, + ) + candidate = get_candidate(runtime, response["token"], "alice") + imported, _layout, details = prepare_apply( + candidate, {"spaces": [], "markers": []}, {}, confirm_missing_content=False, + ) + strips = imported["spaces"][0]["led_strips"] + assert strips[0] == { + "id": "led-a", "points": [[0.1, 0.1], [0.5, 0.1], [0.5, 0.4]], + "marker": "lamp", "active": False, + } + assert strips[1]["marker"] is None + assert candidate["details"]["unbound_led_strips"] == 0 + + +def test_issue_780_full_import_unbinds_a_link_whose_marker_is_absent(tmp_path: Path) -> None: + config = _led_config() + document = _led_document(tmp_path, "full", config=config) + document["payload"]["config"]["markers"] = [] + document["payload"]["layout"] = {} + document["placement_manifest"] = [] + runtime = SimpleNamespace(instance_id="instance-a", import_previews={}) + response = create_preview( + runtime, json.dumps(document).encode(), owner_id="alice", duplicate_policy="skip", + current_config_data={"config": {"spaces": [], "markers": []}, "rev": 0}, + current_layout_data={"layout": {}, "rev": 0}, config_root=tmp_path, + ) + candidate = get_candidate(runtime, response["token"], "alice") + strip = candidate["target_config"]["spaces"][0]["led_strips"][0] + assert strip["marker"] is None and strip["active"] is True + assert strip["points"] == [[0.1, 0.1], [0.5, 0.1], [0.5, 0.4]] + assert candidate["details"]["unbound_led_strips"] == 1 + + +def test_issue_780_space_import_remaps_the_link_through_the_marker_id_map(tmp_path: Path) -> None: + document = _led_document(tmp_path, "space") + # The target already owns a marker with the source id: a coinciding old id + # must never bind the strip to the target's own device. + target = _config() + target["markers"][0]["binding"] = "entity:light.other" + merged, _layout, details = build_space_merge(document, target, {}, "skip") + space = merged["spaces"][-1] + imported_marker = next( + m for m in merged["markers"] + if m.get("space") == details["space_id"] and m.get("binding") == "entity:light.living" + ) + strip = space["led_strips"][0] + assert strip["marker"] == imported_marker["id"] != "lamp" + assert imported_marker["space"] == space["id"] + assert space["led_strips"][1]["marker"] is None + assert details["unbound_led_strips"] == 0 + + +def test_issue_780_skipped_duplicate_marker_leaves_an_unbound_strip(tmp_path: Path) -> None: + document = _led_document(tmp_path, "space") + merged, _layout, details = build_space_merge(document, _config(), {}, "skip") + assert details["skipped"] == 1 + strip = merged["spaces"][-1]["led_strips"][0] + assert strip["marker"] is None and strip["active"] is True + assert strip["points"] == [[0.1, 0.1], [0.5, 0.1], [0.5, 0.4]] + assert details["unbound_led_strips"] == 1 + + +def test_issue_780_virtualised_duplicate_does_not_keep_the_link(tmp_path: Path) -> None: + document = _led_document(tmp_path, "space") + merged, _layout, details = build_space_merge(document, _config(), {}, "virtual") + assert details["virtualized"] >= 1 + assert merged["spaces"][-1]["led_strips"][0]["marker"] is None + assert details["unbound_led_strips"] == 1 + + +def test_issue_780_plan_only_export_keeps_geometry_without_device_links(tmp_path: Path) -> None: + document = _led_document(tmp_path, "space", plan_only=True) + strips = document["payload"]["config"]["spaces"][0]["led_strips"] + assert strips == [ + {"id": "led-a", "points": [[0.1, 0.1], [0.5, 0.1], [0.5, 0.4]], "marker": None, "active": True}, + {"id": "led-b", "points": [[0.2, 0.6], [0.8, 0.6]], "marker": None, "active": True}, + ] + assert "lamp" not in json.dumps(document["payload"]) + # A forged link in a plan-only document has no live marker to point at + # (plan-only carries none): the shared write path unbinds it, nothing leaks. + forged = copy.deepcopy(document) + forged["payload"]["config"]["spaces"][0]["led_strips"][0]["marker"] = "lamp" + parsed = parse_document(json.dumps(forged).encode()) + assert parsed["payload"]["config"]["spaces"][0]["led_strips"][0]["marker"] is None diff --git a/tests_backend/test_ha_websocket.py b/tests_backend/test_ha_websocket.py index cc6ebf1b..953bb2e5 100644 --- a/tests_backend/test_ha_websocket.py +++ b/tests_backend/test_ha_websocket.py @@ -4428,3 +4428,60 @@ async def test_333_optimize_refreshes_the_junction_baseline_cache( "config/set после optimize обязан взять baseline из кэша, " f"а вышло: {baseline_sides}" ) + + +async def test_issue_780_old_client_deleting_a_bound_marker_still_saves( + hass: HomeAssistant, hass_ws_client: WebSocketGenerator, +) -> None: + """A client that does not know LED strips deletes the bound marker: the + write stands, the strip keeps its geometry unbound, and the answer tells a + new client to re-read. A duplicate or foreign link still rejects.""" + await _setup(hass) + client = await hass_ws_client(hass) + space = _space("f1", "r1") + space["led_strips"] = [{ + "id": "led-1", "points": [[0.1, 0.1], [0.6, 0.1], [0.6, 0.5]], + "marker": "lamp", "active": False, + }] + config = { + "spaces": [space, _space("f2", "r2")], + "markers": [{"id": "lamp", "binding": "entity:light.kitchen"}], + "settings": {}, + } + await client.send_json_auto_id({ + "type": "houseplan/config/set", "config": config, "expected_rev": 0, + }) + first = await client.receive_json() + assert first["success"], first + # The empty marker.space was adopted from the strip's space. + assert first["result"]["led_strips"] == {"unbound": 0, "space_adopted": 1} + + old_client = copy.deepcopy(config) + old_client["markers"] = [] + await client.send_json_auto_id({ + "type": "houseplan/config/set", "config": old_client, + "expected_rev": first["result"]["rev"], + }) + second = await client.receive_json() + assert second["success"], second + assert second["result"]["led_strips"] == {"unbound": 1, "space_adopted": 0} + + await client.send_json_auto_id({"type": "houseplan/config/get"}) + stored = (await client.receive_json())["result"] + strip = stored["config"]["spaces"][0]["led_strips"][0] + assert strip == { + "id": "led-1", "points": [[0.1, 0.1], [0.6, 0.1], [0.6, 0.5]], + "marker": None, "active": True, + } + + foreign = copy.deepcopy(config) + foreign["markers"] = [{"id": "lamp", "binding": "entity:light.kitchen", "space": "f2"}] + await client.send_json_auto_id({ + "type": "houseplan/config/set", "config": foreign, + "expected_rev": second["result"]["rev"], + }) + rejected = await client.receive_json() + assert not rejected["success"] + await client.send_json_auto_id({"type": "houseplan/config/get"}) + unchanged = (await client.receive_json())["result"] + assert unchanged["rev"] == second["result"]["rev"], "a rejected write leaves the revision" diff --git a/tests_backend/test_led_strips.py b/tests_backend/test_led_strips.py new file mode 100755 index 00000000..62454720 --- /dev/null +++ b/tests_backend/test_led_strips.py @@ -0,0 +1,279 @@ +"""#780: LED strip storage — schema, write-path normalisation, transfer rules. + +Pure suite (no Home Assistant): ``validation.py`` and ``led_strips.py`` are +loaded by path. The import/export paths that need the HA harness live in +``test_ha_import_export.py``. +""" +from __future__ import annotations + +import copy +import json +from pathlib import Path + +import pytest +import voluptuous as vol + +from pure_imports import load_pure + +_HOUSEPLAN = Path(__file__).resolve().parent.parent / "custom_components" / "houseplan" +v = load_pure("hp_validation_led", _HOUSEPLAN / "validation.py") +led = load_pure("custom_components.houseplan.led_strips", _HOUSEPLAN / "led_strips.py") +support = load_pure("custom_components.houseplan.support_package", _HOUSEPLAN / "support_package.py") +wsm = load_pure("custom_components.houseplan.wall_segment_model", _HOUSEPLAN / "wall_segment_model.py") +PLAN_MODEL_VERSION = 10 + + +def _space(space_id: str, title: str) -> dict: + return { + "id": space_id, "title": title, "view_box": [0, 0, 1, 1], + "rooms": [{"id": f"{space_id}-room", "name": title, "poly": [[0, 0], [1, 0], [1, 1]]}], + } + + +def _config(strips=None, markers=None, *, space_id: str = "ground", extra_spaces=()) -> dict: + """A current (v10) document: built from a v9 one by the real migration.""" + legacy = { + "model_version": PLAN_MODEL_VERSION - 1, + "spaces": [_space(space_id, "Ground"), *(_space(s, s.title()) for s in extra_spaces)], + "markers": markers if markers is not None else [ + {"id": "lamp", "binding": "entity:light.kitchen", "space": space_id}, + ], + } + config = wsm.commit_wall_segment_model(legacy)[0] + if strips is not None: + config["spaces"][0]["led_strips"] = strips + return config + + +def _strip(**extra) -> dict: + return {"id": "led-1", "points": [[0.1, 0.1], [0.4, 0.1], [0.4, 0.3]], "marker": "lamp", **extra} + + +def _check(config: dict) -> dict: + return v.CONFIG_SCHEMA(copy.deepcopy(config)) + + +def _strips_of(config: dict) -> list[dict]: + return config["spaces"][0]["led_strips"] + + +# ---------- AC1: valid geometry is stored ---------- + +def test_open_and_closed_strips_are_stored_exactly() -> None: + closed = {"id": "led-2", "points": [[0, 0], [0.2, 0], [0.2, 0.2], [0, 0]], "marker": None} + checked = _check(_config([_strip(), closed])) + stored = _strips_of(checked) + assert stored[0]["points"] == [[0.1, 0.1], [0.4, 0.1], [0.4, 0.3]] + assert stored[0]["marker"] == "lamp" + assert stored[1]["points"][0] == stored[1]["points"][-1] + assert led.strip_is_closed(stored[1]["points"]) + + +def test_absent_field_equals_no_strips_and_needs_no_version_bump() -> None: + checked = _check(_config()) + assert "led_strips" not in checked["spaces"][0] + assert led.led_strip_counts(checked) == {"led_strips": 0, "led_strips_unbound": 0, "led_strips_hidden": 0} + + +@pytest.mark.parametrize("points, reason", [ + ([[0, 0]], "one point"), + ([[0, 0], [0, 0]], "zero length"), + ([[0, 0], [0, 0], [0, 0]], "zero length, repeated"), + ([[0, 0], [0.2, 0], [0, 0]], "closed with two distinct vertices"), + ([["0", 0], [1, 1]], "string coordinate"), + ([[True, 0], [1, 1]], "boolean coordinate"), + ([[float("nan"), 0], [1, 1]], "NaN"), + ([[float("inf"), 0], [1, 1]], "Infinity"), + ([[0, 0], [6000, 0]], "outside the canvas"), + ([[0, 0, 0], [1, 1]], "three coordinates"), + ("[[0,0],[1,1]]", "not a list"), +]) +def test_invalid_geometry_rejects_the_whole_write(points, reason) -> None: + config = _config([_strip(points=points)]) + before = copy.deepcopy(config) + with pytest.raises(vol.Invalid): + _check(config) + assert config == before, f"{reason}: a rejected write must not be half-applied" + + +def test_point_and_strip_limits_are_50_inclusive() -> None: + fifty = [[i / 100, (i % 2) / 100] for i in range(50)] + assert len(_strips_of(_check(_config([_strip(points=fifty)])))[0]["points"]) == 50 + with pytest.raises(vol.Invalid): + _check(_config([_strip(points=fifty + [[0.9, 0.9]])])) + strips = [{"id": f"led-{i}", "points": [[0, i / 100], [0.1, i / 100]], "marker": None} for i in range(50)] + assert len(_strips_of(_check(_config(strips)))) == 50 + strips.append({"id": "led-50", "points": [[0, 0.9], [0.1, 0.9]], "marker": None}) + with pytest.raises(vol.Invalid): + _check(_config(strips)) + + +def test_hidden_shapes_count_toward_the_strip_limit() -> None: + markers = [{"id": f"m{i}", "binding": "virtual", "space": "ground"} for i in range(51)] + strips = [ + {"id": f"led-{i}", "points": [[0, i / 100], [0.1, i / 100]], "marker": f"m{i}", "active": False} + for i in range(51) + ] + with pytest.raises(vol.Invalid): + _check(_config(strips, markers)) + + +def test_duplicate_strip_id_within_a_space_rejects() -> None: + other = {"id": "led-1", "points": [[0.5, 0.5], [0.6, 0.5]], "marker": None} + with pytest.raises(vol.Invalid, match="unique"): + _check(_config([_strip(), other])) + + +@pytest.mark.parametrize("active", ["true", 1, 0, None]) +def test_active_is_strictly_boolean(active) -> None: + with pytest.raises(vol.Invalid): + _check(_config([_strip(active=active)])) + + +def test_hidden_shape_must_belong_to_a_marker() -> None: + with pytest.raises(vol.Invalid, match="belong to a marker"): + _check(_config([_strip(marker=None, active=False)])) + + +@pytest.mark.parametrize("marker", ["", 5, ["lamp"]]) +def test_marker_reference_must_be_a_non_empty_string_or_null(marker) -> None: + with pytest.raises(vol.Invalid): + _check(_config([_strip(marker=marker)])) + + +# ---------- AC1: duplicates and foreign spaces reject, before any normalisation ---------- + +def test_one_marker_cannot_be_bound_to_two_strips_even_hidden() -> None: + second = {"id": "led-2", "points": [[0.5, 0.5], [0.6, 0.5]], "marker": "lamp", "active": False} + with pytest.raises(vol.Invalid, match="more than one LED strip"): + _check(_config([_strip(), second])) + + +def test_two_links_to_the_same_missing_marker_are_still_a_conflict() -> None: + first = _strip(marker="gone") + second = {"id": "led-2", "points": [[0.5, 0.5], [0.6, 0.5]], "marker": "gone"} + with pytest.raises(vol.Invalid, match="more than one LED strip"): + _check(_config([first, second])) + + +def test_duplicates_across_spaces_are_judged_globally() -> None: + config = _config([_strip()], extra_spaces=("upper",)) + config["spaces"][1]["led_strips"] = [{"id": "led-1", "points": [[0, 0], [1, 0]], "marker": "lamp"}] + with pytest.raises(vol.Invalid, match="more than one LED strip"): + _check(config) + + +def test_marker_of_another_space_rejects() -> None: + markers = [{"id": "lamp", "binding": "entity:light.kitchen", "space": "upper"}] + with pytest.raises(vol.Invalid, match="another space"): + _check(_config([_strip()], markers, extra_spaces=("upper",))) + + +# ---------- §9: normalisation of an old client's write ---------- + +@pytest.mark.parametrize("active", [True, False]) +def test_deleted_marker_leaves_an_unbound_strip_with_its_geometry(active) -> None: + """A client that does not know strips deletes the bound marker: the save stands.""" + config = _config([_strip(active=active)], markers=[]) + report = led.led_strip_link_report(config) + checked = _check(config) + strip = _strips_of(checked)[0] + assert strip["id"] == "led-1" + assert strip["points"] == [[0.1, 0.1], [0.4, 0.1], [0.4, 0.3]] + assert strip["marker"] is None + assert strip["active"] is True + assert report == {"unbound": 1, "space_adopted": 0} + + +def test_tombstoned_marker_is_not_live() -> None: + markers = [{"id": "lamp", "binding": "entity:light.kitchen", "space": "ground", "removed": True}] + strip = _strips_of(_check(_config([_strip()], markers)))[0] + assert strip["marker"] is None and strip["active"] is True + + +@pytest.mark.parametrize("space", ["absent", None, ""]) +def test_empty_marker_space_adopts_the_strip_space(space) -> None: + marker = {"id": "lamp", "binding": "entity:light.kitchen"} + if space != "absent": + marker["space"] = space + config = _config([_strip()], [marker]) + assert led.led_strip_link_report(config) == {"unbound": 0, "space_adopted": 1} + checked = _check(config) + assert checked["markers"][0]["space"] == "ground" + assert _strips_of(checked)[0]["marker"] == "lamp" + + +def test_normalisation_is_idempotent() -> None: + config = _config([_strip(active=False)], markers=[]) + once = _check(config) + twice = _check(once) + assert once == twice + assert led.led_strip_link_report(once) == {"unbound": 0, "space_adopted": 0} + + +def test_unavailable_entity_is_not_a_missing_marker() -> None: + """Temporary HA unavailability is a state, never a config fact: the link stays.""" + strip = _strips_of(_check(_config([_strip()])))[0] + assert strip["marker"] == "lamp" + + +def test_points_carry_only_json_noise_cleanup_not_a_lattice_snap() -> None: + face = 0.4123456789 # a physical wall face, off the plan lattice + strip = _strip(points=[[0.1, face], [0.30000000000000004, face]]) + stored = _strips_of(_check(_config([strip])))[0]["points"] + assert stored == [[0.1, 0.412345679], [0.3, 0.412345679]] + + +# ---------- transfer rules ---------- + +def test_plan_only_projection_keeps_geometry_and_drops_every_device_link() -> None: + space = {"led_strips": [ + _strip(), + {"id": "led-2", "points": [[0, 0], [1, 0]], "marker": "fan", "active": False, "future": "x"}, + ]} + projected = led.plan_only_strips(space) + assert projected == [ + {"id": "led-1", "points": [[0.1, 0.1], [0.4, 0.1], [0.4, 0.3]], "marker": None, "active": True}, + {"id": "led-2", "points": [[0, 0], [1, 0]], "marker": None, "active": True}, + ] + assert "lamp" not in json.dumps(projected) and "fan" not in json.dumps(projected) + assert led.plan_only_strips({}) is None + + +def test_transfer_remaps_links_and_unbinds_what_did_not_travel() -> None: + space = {"led_strips": [ + _strip(), + {"id": "led-2", "points": [[0, 0], [1, 0]], "marker": "skipped", "active": False}, + {"id": "led-3", "points": [[0, 1], [1, 1]], "marker": None}, + ]} + unbound = led.unbind_strips(space, remap={"lamp": "marker_lamp_2"}) + strips = space["led_strips"] + assert strips[0]["marker"] == "marker_lamp_2" + assert strips[1] == {"id": "led-2", "points": [[0, 0], [1, 0]], "marker": None, "active": True} + assert strips[2]["marker"] is None + assert unbound == 1 + + +def test_transfer_never_binds_by_a_coinciding_old_id() -> None: + """The target has its own marker 'lamp': without a remap entry the strip unbinds.""" + space = {"led_strips": [_strip()]} + assert led.unbind_strips(space, remap={}) == 1 + assert space["led_strips"][0]["marker"] is None + + +# ---------- diagnostics privacy ---------- + +def test_support_summary_counts_strips_without_coordinates_or_ids() -> None: + config = _config([ + _strip(), + {"id": "led-secret-id", "points": [[0.123, 0.456], [0.789, 0.456]], "marker": None}, + ]) + config["markers"].append({"id": "hall", "binding": "entity:light.hall", "space": "ground"}) + config["spaces"][0]["led_strips"].append( + {"id": "led-3", "points": [[0, 0], [1, 0]], "marker": "hall", "active": False}, + ) + summary = support._summary(config, {}) + assert summary["led_strips"] == {"total": 3, "unbound": 1, "hidden": 1} + text = json.dumps(summary) + for secret in ("led-secret-id", "0.123", "0.789", "light.kitchen", "light.hall"): + assert secret not in text