feat(backend): LED strips are stored, normalised and transferred (#780)

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
This commit is contained in:
claude[bot]
2026-10-02 20:54:49 +03:00
committed by Codex
parent 9fa5efbdac
commit 870237f3fb
13 changed files with 875 additions and 1 deletions
@@ -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",))
@@ -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:
+287
View File
@@ -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}
@@ -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":
@@ -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,
)
+10 -1
View File
@@ -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 ----------------
+1
View File
@@ -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
+27
View File
@@ -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": "<function _strict_bool>",
"required": false
},
"config.spaces[].led_strips[].id": {
"opaque": "<function _strip_id>",
"required": true
},
"config.spaces[].led_strips[].marker": {
"default": {
"factory": "NoneType"
},
"opaque": "<function _marker_ref>",
"required": false
},
"config.spaces[].led_strips[].points": {
"required": true
},
"config.spaces[].led_strips[].points[][]": {
"opaque": "<function _coordinate>"
},
"config.spaces[].open_spans": {
"required": false
},
+14
View File
@@ -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<T>(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']);
+14
View File
@@ -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
}
+117
View File
@@ -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
+57
View File
@@ -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"
+279
View File
@@ -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