Merge pull request #199 from mbillow/claude/multi-device-subdevice-patterns-1xrtuh

Support multi-indoor-unit ("2-in-1") systems (#177)
This commit is contained in:
Marc Billow
2026-07-29 14:42:27 -05:00
committed by GitHub
34 changed files with 5373 additions and 128 deletions
+96 -3
View File
@@ -10,7 +10,9 @@ description: >-
OCF-standard vs vendor hrefs, the diagnostic/config/normal entity taxonomy,
preferring dynamic (device-reported) select options over hardcoded lists,
ensuring every href is bound or ignored, and locking it in with a fixture +
golden + test.
golden + test. Also covers multi-unit ("composite") appliances that expose
several logical indoor units over one IP — triaging a missing second unit,
and why registry hrefs stay canonical rather than indexed.
---
# Adding device support
@@ -25,11 +27,25 @@ dump into coverage.
A user's diagnostics download (`config_entry-localthings-*.json`) has, under
`data`:
- `resources`: `{href: rep}` — the parsed `/device/0` snapshot. **This is the
source of truth**, not code comments.
source of truth**, not code comments. On a multi-unit appliance this is the
unit the config entry connects to and *only* that unit; siblings report
their own (see below).
- `unbound_hrefs`: resources that bound to no capability. The
"incomplete capability coverage" repair fires whenever this is **non-empty or
the device type is unrecognized** (`coordinator._update_coverage_gap_issue`).
Multi-unit appliances (one IP, one DTLS session, several logical indoor
units — issue #177) add four more, all absent/empty on an ordinary device:
- `sub_units`: one entry per materialized sibling — `kind`/`key`/`seed_path`,
its own `model`, its bound hrefs, and its own `resources`.
- `sub_units_skipped`: candidates whose seed answered but that produced no
live primary state, with the reps the gate actually judged. An unused
SmartThings slot lands here, not in `sub_units`.
- `sub_unit_probes`: `{seed_href: found}` for every seed attempted — tells
"checked, nothing there" apart from "never checked".
- `multidevice`: `/multidevice/vs/0`'s rep if the board answers it. Its
`numofsubdevice` is a corroborating count, not a gate.
Goal: make `unbound_hrefs` empty by **binding** the useful resources and
**ignoring** the noise — and surface every genuinely useful sensor/select/switch
along the way.
@@ -61,6 +77,12 @@ print('state_keys:', sorted(state))
`exists_fn` and produces the final entity values. Use the same routine to
regenerate a golden.
**A sibling unit's block runs through this unchanged.** `sub_units[i].resources`
(and `sub_units_skipped[i].resources`) are keyed by *canonical* hrefs —
`/mode/vs/0`, never the `/mode/vs/1` or `/<uuid>/mode/vs/0` that unit actually
answers on — precisely so you can paste one into `resources` above and read the
result exactly like the master's. No de-indexing by hand.
## 3. Route the device to a registry — add a row, never a branch
If detection returns `None`, the device falls back to common capabilities and
@@ -192,6 +214,15 @@ sub-polled between summary polls. Pick descriptor types from `entities.py`
as a gap for a human, or ignore it with a documented reason — never invent an
entity on a hunch (`ignored.py`'s rule).
This is a rule about **writes and entities**, not about reading. A speculative
`GET` of an href a dump doesn't contain is fine and the codebase already relies
on it: `read_identity` reads `/oic/p`, `/oic/d` and `/oic/res`, and
`subunits.enumerate_sub_units` probes `/device/<n>`, `/<uuid>/device/0` and
`/multidevice/vs/0` on every device. A RETRIEVE is non-mutating and a 4.04 is
tolerated everywhere in that path, so the cost of a wrong guess is one wasted
round trip. Guessing a *write* against live hardware is the thing this rule
forbids — as is materializing an entity from a field you can't explain.
## 6. Select options: read them from the device, don't hardcode
A `SelectDesc`'s `options` should come from the device's own advertised list
@@ -275,6 +306,17 @@ friendlier href**.
ignored because washers bind it. When only one family should ignore an href
that another binds, scope the ignore to that family's registry.
- **Registry hrefs are always canonical — never index or prefix one.** On a
multi-unit appliance, `unbound_hrefs` reports the *real* href a gap was seen
on, so a sibling's gap shows up as `/foo/vs/1` or
`/<uuid>/foo/vs/0`. Do **not** write `Capability(href='/foo/vs/1')` for it.
Binding runs against each unit's canonical view, so an indexed or prefixed
href in a registry matches nothing on any device and fails silently — no
error, no entity, and the gap stays open. Fix it on the `/foo/vs/0` form and
every unit gets it at once. (`registry/subunits.py` owns the canonical ⇄
actual translation; nothing under `capabilities/` or `by_type/` should ever
mention a unit index.)
## 9. Reuse before writing new code
Check `common.py` (generic OCF: power, energy, alarms, water) and `laundry.py`
@@ -289,8 +331,26 @@ shared module rather than copying.
1. Add a **scrubbed** fixture `tests/fixtures/<type>_device.json`
(`{"device0": [ {devcol rep}, {href, rep}, ... ]}`) — replace serials, MACs,
and other PII with placeholders.
A multi-unit dump (issue #177) may carry three more top-level keys, all
optional and defaulted for every other fixture — load them with
`conftest._load_device_full` rather than `_load_device`:
- `oic_res`: the raw `/oic/res` link array, which is what enumeration reads
to find `/device/<n>` siblings.
- `seeds`: `{seed_href: raw_batch_list}` — each sibling's own collection
response, in the same `[devcol rep, {href, rep}, ...]` shape as `device0`.
- `probes`: `{href: rep}` for plain Property-map resources belonging to no
batch (e.g. a hand-read `/multidevice/vs/0`).
Add a `seeds_note` saying which parts are verbatim captures and which were
constructed. A fixture that quietly mixes the two is worse than no fixture:
the whole point of the corpus is that it records what hardware actually did.
2. Generate `tests/fixtures/golden/<type>.json` (`{"state_keys": [...]}`) with
the harness in §2.
the harness in §2. A multi-unit fixture's golden carries a sibling's keys
under a prefix (`unit1_climate`, `sub_<uuid>_climate`) alongside the
unprefixed master keys — that's the entity-ID namespacing, not a bug.
The master's keys are unprefixed *by design* and must never gain one:
that's what keeps every pre-#177 device's `unique_id` stable.
3. Add the type to `test_golden_regression.py` and write a
`test_<type>_capabilities.py` asserting **zero unbound hrefs** and that the
expected entities exist (and any misleading ones are gated).
@@ -303,7 +363,40 @@ The new fixture is picked up automatically by the corpus-wide checks (the
board token fails the build rather than silently mistyping someone's
appliance.
## 11. Triage: "one of my units is missing"
For an appliance that exposes several logical indoor units over one IP —
a 2-in-1 air conditioner, plausibly a multi-drum washer (#19). Work down
the dump in this order; each step rules out a different cause.
1. **`sub_unit_probes`** — did we even look? Every seed attempted appears
here with what it returned. An absent seed means enumeration never tried
that path; a `false` means it tried and got nothing.
2. **`sub_units_skipped`** — did we find it and reject it? A candidate lands
here when its seed answered but it produced no *primary* (non-diagnostic)
entity with a populated value. Its `resources` block holds the exact reps
the gate judged, so you can check the call yourself. If every
power/mode/temperature rep is `{}`, the unit is an unused slot and the
skip is correct. If they're populated, the gate is wrong — that's a bug
worth a fixture.
3. **`multidevice.numofsubdevice`** — the board's own count, where it
reports one. Disagreement with `len(sub_units) + 1` is a strong hint,
not proof; only one board family is known to expose it.
4. **Which pattern is this board?** `identity.resources['/oic/res']` listing
`/device/1`, `/device/2` means indexed siblings. `resources['/subdevices/
vs/0']` carrying a `subdeviceIdList` means a UUID-prefixed tree, and that
same UUID usually shows up as an href prefix in `/oic/res` too. Neither
present, on a device the owner insists has two units, is the interesting
case — that's a third mechanism and needs a new dump, not a code guess.
Two things that are *not* the fix: adding a capability for an indexed href
(see §8), and loosening the liveness gate to "any populated entity" — a
rejected slot routinely reports a non-`None` *diagnostic* value off an empty
resource, which is exactly what the primary-entity filter exists to ignore.
## Key files
- `registry/subunits.py` — `SubUnit`, enumeration, canonical ⇄ actual href
translation, and the materialization gate for multi-unit appliances.
- `registry/discovery.py` — `discover()`, unbound reporting, pattern caps.
- `registry/capability.py`, `registry/entities.py` — the `Capability` and
descriptor shapes (`rt_filter`, `match_fn`, `exists_fn`, `rep_fn`, `write_fn`).
+11
View File
@@ -185,6 +185,17 @@ Samsung's firmware occasionally drops the DTLS session briefly — this is norma
If reconnects become persistent (more than a handful per minute), something's actually wrong. Check the appliance's Wi-Fi link first, then look for a competing DTLS client on the LAN — only one active session per appliance is allowed at a time.
### Multi-indoor-unit ("2-in-1") air conditioner systems
Some Samsung installs run more than one indoor unit off a single outdoor unit, all reachable over the *one* IP/DTLS session your config entry connects to (a floor-standing + wall-mounted 2-in-1 is a common shape). The integration discovers any sibling units automatically, once, right after the first successful poll — there's nothing to configure. Each discovered unit gets its own HA device (linked to the main one via "via device") and its own `climate` card, so it lands in its own room in the dashboard instead of being invisible or mixed into the master unit's state.
Two on-the-wire shapes are supported, both keyed off what the appliance itself reports:
- **Indexed siblings** — the device answers a `/device/1`, `/device/2`, ... collection alongside its own `/device/0`, mirroring every resource at that index.
- **UUID-prefixed tree** — the device reports a sibling's id in `x.com.samsung.da.subdeviceIdList`, and that id doubles as a literal href prefix for the sibling's own resource tree.
A candidate that answers but never produces any real, user-facing state (an unused slot some installs report alongside a genuine second unit) is silently skipped rather than turned into a phantom entity — check diagnostics' `sub_units`/`sub_units_skipped` blocks if a unit you expect to see isn't showing up, and file an issue with that diagnostics download attached.
---
## Contributing
+31 -11
View File
@@ -254,8 +254,7 @@ class LocalThingsClimate(LocalThingsEntity, ClimateEntity):
def _legacy_convenient(self) -> dict:
"""A /mode/convenient/vs/0-shaped rep built from the Comode_* token in
/mode/vs/0's options, for boards that have no convenient resource."""
options = (self.coordinator.resource(MODE_HREF) or {}).get(
'x.com.samsung.da.options') or []
options = self._rep(MODE_HREF).get('x.com.samsung.da.options') or []
for option in options:
if isinstance(option, str) and option.startswith('Comode_'):
return {_MODES_FIELD: [option.split('_', 1)[1]],
@@ -268,18 +267,26 @@ class LocalThingsClimate(LocalThingsEntity, ClimateEntity):
Delegates the board-generation test to is_legacy_board (the same
test capabilities/airconditioner.py's token entities are gated on)
instead of re-implementing it. Uses last_resources rather than a
two-key presence dict built from coordinator.resource()'s truthiness
-- resource() collapses "href absent" and "href present with an
empty {} rep" to the same falsy value, while is_legacy_board (and
discover()'s own binding) test key membership, not truthiness. A
instead of re-implementing it. Uses self._resources (this unit's own
canonical view, issue #177 -- see LocalThingsEntity._resources)
rather than a two-key presence dict built from coordinator.resource()'s
truthiness -- resource() collapses "href absent" and "href present
with an empty {} rep" to the same falsy value, while is_legacy_board
(and discover()'s own binding) test key membership, not truthiness. A
presence dict built from truthiness alone would disagree with the
token entities on a board reporting a genuinely empty /airflow/vs/0,
silently reintroducing the drift this delegation exists to prevent.
Reads the actual href through self._rep rather than
coordinator.resource() directly -- on a sub-unit (a legacy-board
sibling has its own /airflow/vs/1, or /<id>/airflow/vs/0), the
canonical AIRFLOW_HREF must be translated through this bound
entity's own sub_unit first, exactly like every other sibling read
below.
"""
if not is_legacy_board(self.coordinator.last_resources):
if not is_legacy_board(self._resources):
return {}
return self.coordinator.resource(AIRFLOW_HREF) or {}
return self._rep(AIRFLOW_HREF)
def _legacy_preset(self) -> bool:
"""Whether presets come from the Comode_* token rather than a resource.
@@ -288,12 +295,25 @@ class LocalThingsClimate(LocalThingsEntity, ClimateEntity):
rep being empty alone: newer boards carry Comode tokens too, so a
momentarily empty /mode/convenient/vs/0 there must not silently switch
the preset read (and write) over to the token path.
Deliberately reads the *raw* href (translated through this bound
entity's own sub_unit, not through self._rep) rather than going
through _rep's own CONVENIENT_HREF fallback branch -- that fallback
is exactly the legacy_convenient() rep this method is deciding
whether to use, so routing through it here would make the resource
never look empty and this always resolve to the wrong side.
"""
return (not self.coordinator.resource(CONVENIENT_HREF)
convenient_href = self._bound.sub_unit.to_actual(CONVENIENT_HREF)
return (not self.coordinator.resource(convenient_href)
and bool(self._legacy_airflow()))
def _rep(self, href: str) -> dict:
rep = self.coordinator.resource(href) or {}
"""`href` is one of this module's canonical HREF_* constants --
translated through this bound entity's own sub_unit (issue #177) to
the real, on-the-wire href before the single-href cache lookup
(identity for MAIN, so a device with no sub-units reads exactly the
href it always did)."""
rep = self.coordinator.resource(self._bound.sub_unit.to_actual(href)) or {}
if not rep and href == CONVENIENT_HREF and self._legacy_airflow():
return self._legacy_convenient()
return rep
+288 -16
View File
@@ -30,10 +30,14 @@ from .registry.capabilities.common import (
remote_control_enabled,
remote_control_required_for_write,
)
from .registry.discovery import discover, BoundEntity
from .registry.discovery import BoundEntity
from .registry import CAPABILITIES
from .registry.adapter import flatten
from .registry.identity import read_identity, DeviceIdentity
from .registry.subunits import (
MAIN, SubUnit, canonical_view, discover_partitioned, enumerate_sub_units,
normalize_seed_batch,
)
from .observe import ObserveManager, MODE_OBSERVE, MODE_POLL, GRACE_PERIOD_S
from .const import (
@@ -161,6 +165,39 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self._identity: DeviceIdentity | None = None
self._discovered = False
self.bound = []
# Sibling indoor units discovered on this connection (issue #177) --
# candidates set once, at first discovery, by
# _enumerate_sub_units_blocking; narrowed by _run_discovery to the
# ones that actually produced live primary state (see
# subunits.discover_partitioned). MAIN itself is never in this list
# (see subunits.canonical_view's docstring for why that's safe):
# it's the *other* units sharing this DTLS session, if any.
self.sub_units: list[SubUnit] = []
# Candidates _run_discovery's gate rejected (an unused SmartThings
# slot that still answers its seed, e.g. HJcom's /device/2) --
# surfaced in diagnostics alongside the materialized ones so a
# report shows what was found and why it didn't become an entity.
self._skipped_sub_units: list = []
# Those rejected candidates' raw reps, kept aside for diagnostics
# only (see _live_unit_resources). They are deliberately not in the
# state cache: nothing polls them again, so anything applied there
# would sit frozen at its first-discovery value while looking as
# live as every other href in `last_resources`.
self._skipped_sub_unit_resources: dict[str, dict] = {}
# /multidevice/vs/0's rep, if this board answers it -- a plain unit
# count that corroborates the liveness gate without deciding it.
# Deliberately outside `resources`; see _enumerate_sub_units_blocking.
self._multidevice: dict = {}
# What each sub-unit probe found, keyed by the seed href attempted --
# surfaced in diagnostics so a report can tell "checked, nothing
# there" apart from "never checked" (the same posture the
# speculative-probe code this replaced documented in identity.py).
self._sub_unit_probes: dict[str, bool] = {}
# canonical_resources() memo, keyed by (sub_unit.kind, sub_unit.key).
# Invalidated in _on_cache_changed -- climate.py reads this on every
# property access (is_legacy_board and friends), so it must not
# rebuild an O(hrefs) view from scratch on every single property.
self._canonical_cache: dict[tuple[str, str], dict] = {}
self._cache = StateCache(_NoOpDescriptor())
self._cache.set_on_change(self._on_cache_changed)
self._observe = ObserveManager(self._cache, logger=self._log)
@@ -197,6 +234,69 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
direct O(1) cache lookup."""
return self._cache.get(href) or {}
def canonical_resources(self, sub_unit: SubUnit) -> dict[str, dict]:
"""`sub_unit`'s own view of the live snapshot, rewritten into the
canonical hrefs (issue #177) the registry/platforms are written
against -- see subunits.canonical_view. A platform property that
needs the *whole* resources dict (as opposed to one href via
`resource()`/`last_resources.get(href)`) must use this instead of
`last_resources`, or a sibling unit's own `/mode/vs/1` would leak
into MAIN's canonical `/mode/vs/0` view (or vice versa) under
exists_fn/is_legacy_board-style checks that scan the whole dict.
Memoized per cache generation: climate.py calls this on every
property read (is_legacy_board and friends), and building it is
O(hrefs) -- _on_cache_changed clears the memo whenever the
snapshot actually changes, not on every property access.
"""
view_key = (sub_unit.kind, sub_unit.key)
cached = self._canonical_cache.get(view_key)
if cached is not None:
return cached
view = canonical_view(sub_unit, self._cache.snapshot(), self.sub_units)
self._canonical_cache[view_key] = view
return view
def device_info_for(self, sub_unit: SubUnit) -> DeviceInfo:
"""DeviceInfo for one logical unit sharing this connection (issue
#177) -- the master's own (unchanged) device_info for MAIN, or a
linked child device for a discovered sub-unit.
Identifiers derive from the *master's* serial (device_serial) plus
this unit's stable key, never from whatever serial the sub-unit
itself reports (or fails to) -- deterministic across reconnects
whether or not this unit's own identity resource
(/information/vs/<n>, or /<id>/information/vs/0) answered on the
poll that first created the HA device. `serial_number` is set from
that resource when present anyway -- it's informational, not an
identifier.
"""
if sub_unit.kind == 'main':
return self.device_info
info = self.canonical_resources(sub_unit).get('/information/vs/0', {})
model_num = info.get('x.com.samsung.da.modelNum', '')
model = model_num.split('|', 1)[0] if model_num else ''
serial = info.get('x.com.samsung.da.serialNum') or None
base_name = self.device_info.get('name') or 'Samsung Appliance'
if model:
label = model.replace('_', ' ').title()
else:
# This poll never got (or never will get) the sub-unit's own
# identity resource -- fall back to a generic per-unit label
# rather than leaving the device unnamed. 'Unit <n>' only makes
# sense for an indexed unit (the key is a small ordinal);
# jhkwon19-pattern (prefixed) units are never more than one per
# connection today, so there's no ordinal to show.
label = f'Unit {sub_unit.key}' if sub_unit.kind == 'indexed' else 'Secondary Unit'
return DeviceInfo(
identifiers={(DOMAIN, f"{self.device_serial}_{sub_unit.key}")},
via_device=(DOMAIN, self.device_serial),
name=f"{base_name} {label}",
manufacturer=self.device_info.get('manufacturer') or 'Samsung',
model=model or None,
serial_number=serial,
)
@property
def observe_mode(self) -> str:
return self._observe.mode
@@ -245,6 +345,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
into a single push instead of one hass.add_job per href."""
if not changed:
return
self._canonical_cache.clear()
with self._push_pending_lock:
if self._push_pending:
return
@@ -292,7 +393,33 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
body = cbor2.loads(payload)
except Exception as e:
raise RuntimeError(f"poll cbor decode: {e}") from e
return parse_device0_batch(body) if isinstance(body, list) else {}
result = parse_device0_batch(body) if isinstance(body, list) else {}
# Refresh every already-enumerated sibling unit's seed collection on
# this same summary poll (issue #177) -- without this, a sub-unit's
# climate card would show only its enumeration-time snapshot forever.
for sub_unit in self.sub_units:
result.update(self._poll_sub_unit_seed(sub_unit))
return result
def _poll_sub_unit_seed(self, sub_unit: SubUnit) -> dict[str, dict]:
"""GET one sub-unit's seed Collection and return its batch,
normalized to real hrefs. A sibling failing to answer is a debug
log, never a failed poll -- the master unit must not go unavailable
because a sibling timed out or dropped off (e.g. HJcom's
/device/2, a SmartThings-unused component that may not always
respond). Blocking -- called from _poll_once, already in executor."""
sess = self._session
if sess is None:
return {}
try:
code, payload = sess.get(list(sub_unit.seed_path), timeout=10.0)
if code == 0x45 and payload:
body = cbor2.loads(payload)
if isinstance(body, list):
return normalize_seed_batch(sub_unit, parse_device0_batch(body))
except Exception as e:
self._log.debug("sub-unit %s seed poll failed: %s", sub_unit.key, e)
return {}
def _poll_hrefs_blocking(self, hrefs: list[str]) -> dict[str, dict]:
"""GET individual hrefs sequentially. Does not reconnect on failure. Blocking."""
@@ -351,6 +478,71 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
# Discovery (runs once on first successful poll)
# ------------------------------------------------------------------
def _enumerate_sub_units_blocking(self, resources: dict[str, dict]) -> dict[str, dict]:
"""One-time (first discovery only) probe for sibling indoor units
sharing this connection (issue #177) -- see
registry.subunits.enumerate_sub_units for the two detection
patterns. Blocking -- runs in executor, under the session lock
(shares the same DTLS session _poll_once just used this cycle).
Sets self.sub_units to every *candidate* the probes turned up
(self._sub_unit_probes as a side effect too) and returns `resources`
merged with whatever each candidate's seed returned, so this cycle's
_run_discovery sees every candidate's state without a second poll
round trip. `_run_discovery` is what narrows self.sub_units down to
the ones that are actually live (see discover_partitioned) -- this
method doesn't know how to tell an unused SmartThings slot (HJcom's
/device/2) from a real sibling, only that something answered.
"""
if self._session is None:
self._connect_session()
sess = self._session
if sess is None:
return resources
oic_res = self._identity.raw.get('/oic/res', []) if self._identity else []
probes: dict[str, bool] = {}
sub_units, extra = enumerate_sub_units(
sess, resources, oic_res,
probe_log=lambda href, found: probes.__setitem__(href, found),
)
self.sub_units = sub_units
self._sub_unit_probes = probes
# /multidevice/vs/0 is corroborating metadata, not appliance state,
# and it is probed on *every* device -- so it must not join the
# returned resources dict. Two things go wrong if it does. It would
# reach discovery on families whose registry doesn't ignore that
# href (only the AC one does), binding to nothing and raising a
# spurious "incomplete capability coverage" repair for every washer
# or fridge whose firmware happens to answer it. And it is fetched
# once here and never polled again, so applying it to the state
# cache would freeze it there exactly like a rejected candidate's
# reps (see _live_unit_resources). Kept aside for diagnostics and
# for the numofsubdevice cross-check in _run_discovery instead.
self._multidevice = extra.pop('/multidevice/vs/0', {})
return {**resources, **extra}
def _live_unit_resources(self, resources: dict[str, dict]) -> dict[str, dict]:
"""`resources` minus every href belonging to a candidate sub-unit the
liveness gate rejected (issue #177).
Called once, between _run_discovery and the first cache apply, so a
rejected slot's reps are seen by the gate and then dropped rather
than frozen into the cache forever -- see the call site. The reps
themselves are kept in _skipped_sub_unit_resources for diagnostics,
which is the only thing that still wants them.
"""
if not self._skipped_sub_units:
return resources
kept: dict[str, dict] = {}
skipped: dict[str, dict] = {}
for href, rep in resources.items():
bucket = skipped if any(
skip.sub_unit.owns(href) for skip in self._skipped_sub_units
) else kept
bucket[href] = rep
self._skipped_sub_unit_resources = skipped
return kept
def _run_discovery(self, resources: dict[str, dict]) -> None:
# Reported for diagnostics only -- it names the firmware generation
# ('7.0 Air conditioner' is Tizen Lite), which is useful when triaging
@@ -360,7 +552,6 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
.get('swVersionInfo', {})
.get('oneUiVersion', ''))
info = resources.get('/information/vs/0', {})
reg = resolve_registry(resources)
unbound: list[str] = []
hot, warm = set(), set()
@@ -372,11 +563,53 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
model_num = info.get('x.com.samsung.da.modelNum', '')
description = info.get('x.com.samsung.da.description', '')
if reg is not None:
self._log.debug("device type: %s (modelNum=%r)", reg.name, model_num)
bound = discover(resources, reg.capabilities, reg.pattern_capabilities,
log=unbound.append, tier_log=_tier_log)
self.device_type_name = reg.name
# Partitioned discovery (issue #177): the main pass binds every href
# owned by no sub-unit; one further pass per *candidate* sub-unit
# binds its own canonical view, resolving its own device type from
# its own /information/vs/0 when it reports one and falling back to
# the master's registry otherwise. See subunits.discover_partitioned
# -- it also gates each candidate down to whether it actually
# produced live primary state (HJcom's /device/2, an unused
# SmartThings slot, answers its seed but never does), so
# self.sub_units below is narrowed to the ones that passed, not
# every candidate _enumerate_sub_units_blocking found. For a device
# with no candidates (self.sub_units == []) this is exactly the
# single discover() call this method used to make.
bound, device_type_name, materialized, skipped = discover_partitioned(
resources, self.sub_units, resolve_registry, CAPABILITIES,
log=unbound.append, tier_log=_tier_log,
)
self.sub_units = materialized
self._skipped_sub_units = skipped
for skip in skipped:
self._log.info(
"sub-unit %s (%s) answered its seed but produced no live "
"primary state; not materialized (hrefs=%s)",
skip.sub_unit.key, skip.sub_unit.kind, list(skip.hrefs),
)
# Corroborating signal, not a gate (DESIGN-177.md section 4):
# /multidevice/vs/0's numofsubdevice is a plain count HJcom's board
# reports independently of the liveness gate above. Log, don't
# raise, on a disagreement -- only this one board family is known to
# expose the resource at all, so a mismatch is a "look into this"
# signal for triage, not proof either side is wrong.
numofsubdevice = self._multidevice.get(
'x.com.samsung.da.numofsubdevice')
if numofsubdevice is not None:
try:
reported = int(numofsubdevice)
except (TypeError, ValueError):
reported = None
unit_count = len(materialized) + 1 # +1 for the master itself
if reported is not None and reported != unit_count:
self._log.debug(
"/multidevice/vs/0 reports numofsubdevice=%r but %d "
"unit(s) materialized (including the master)",
numofsubdevice, unit_count,
)
if device_type_name is not None:
self._log.debug("device type: %s (modelNum=%r)", device_type_name, model_num)
else:
# Both fields: detection reads each of them (board token, then
# consumer-model code), and this line is what a user pastes into
@@ -385,8 +618,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"unknown device type modelNum=%r description=%r; using common caps",
model_num, description,
)
bound = discover(resources, CAPABILITIES, log=unbound.append, tier_log=_tier_log)
self.device_type_name = None
self.device_type_name = device_type_name
self.bound = bound
self._unbound_hrefs = unbound
@@ -396,7 +628,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self.device_serial = serial
ident = self._identity
device_type = reg.name.replace('_', ' ').title() if reg else 'Appliance'
device_type = device_type_name.replace('_', ' ').title() if device_type_name else 'Appliance'
model = model_num.split('|', 1)[0] if model_num else (ident.model if ident else '')
name = f"Samsung {device_type} ({model})" if model else f"Samsung {device_type}"
mfr = (ident.manufacturer if ident else '') or 'Samsung'
@@ -407,15 +639,16 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
manufacturer=mfr,
model=model,
)
self._update_coverage_gap_issue(reg is None, unbound, name)
self._update_coverage_gap_issue(device_type_name is None, unbound, name)
self._hot_hrefs = sorted(hot)
self._warm_hrefs = sorted(warm)
self._discovered = True
self._log.info(
"discovered %d entities (serial=%s) hot=%s warm=%s",
"discovered %d entities (serial=%s) hot=%s warm=%s sub_units=%s",
len(bound), serial, self._hot_hrefs, self._warm_hrefs,
[su.key for su in self.sub_units],
)
def _update_coverage_gap_issue(
@@ -568,7 +801,37 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self._observe.downgrade_to_poll()
just_downgraded_from_observe = True
if not self._discovered:
# One-time (issue #177): find out whether this connection has
# sibling indoor units before the first discovery pass, and fold
# their seed resources into this cycle's snapshot so discovery
# sees every unit's state on the very first poll rather than
# waiting a cycle. Runs under its own session-lock scope (the
# poll above already released the lock) since it shares the same
# DTLS session.
async with self._session_lock:
resources = await self.hass.async_add_executor_job(
self._enumerate_sub_units_blocking, resources
)
source = 'sweep' if self._discovered else 'poll'
first_cycle = not self._discovered
if first_cycle:
# Discovery runs *before* the apply loop below, not after it, so
# a rejected candidate's resources never reach the state cache
# at all (issue #177). Enumeration has to fetch every candidate's
# seed to evaluate the liveness gate, but only the units that
# pass it are ever polled again -- applying the rest would freeze
# ~14 hrefs per rejected slot into the cache on this one cycle
# and leave them there forever, indistinguishable from live
# state in `last_resources` and in the diagnostics dump built
# from it. StateCache has no eviction, so the only way to keep
# them out is to not put them in. Safe to reorder: _run_discovery
# reads the dict passed to it and never the cache, and
# log_sweep_discrepancies below can't fire on a first cycle
# (observe mode is only ever attempted after discovery).
self._run_discovery(resources)
resources = self._live_unit_resources(resources)
sweep_mismatch = False
if self._observe.mode == MODE_OBSERVE:
# A sweep/cache mismatch never tears down a still-live OBSERVE
@@ -584,8 +847,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
for href, rep in resources.items():
self._observe.apply(href, rep, source=source)
if not self._discovered:
self._run_discovery(resources)
if first_cycle:
await self._attempt_observe_mode()
elif just_downgraded_from_observe:
await self._attempt_observe_mode()
@@ -671,7 +933,17 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
# state until the next real read of that resource -- the 20-60s lag
# in issues #17/#53, which survived the earlier optimistic-apply fix
# (issue #27) because that fix applied to the wrong href too.
write_href = '/' + '/'.join(path_segs)
#
# write_fn's path_segs are canonical (issue #177) -- a sub-unit's
# ClimateDesc is bound to its own *actual* /mode/vs/1 (or
# /<id>/mode/vs/0) href, but _climate_write only knows the canonical
# sibling hrefs (e.g. ['power', 'vs', '0']). Translate through this
# bound entity's own sub_unit so the optimistic apply, the settle
# guard and the POST below all target that unit's real resource --
# to_actual is the identity transform for MAIN, so a device with no
# sub-units writes exactly where it always did.
write_href = bound_entity.sub_unit.to_actual('/' + '/'.join(path_segs))
path_segs = [s for s in write_href.strip('/').split('/') if s]
# Apply the write optimistically before starting the settle guard,
# not after -- mark_write_pending gates every source (poll, sweep,
+89 -1
View File
@@ -18,6 +18,7 @@ from homeassistant.loader import async_get_integration
from .const import DOMAIN
from .coordinator import LocalThingsCoordinator
from .registry.redact import redact_resources
from .registry.subunits import MAIN
async def async_get_config_entry_diagnostics(
@@ -39,6 +40,33 @@ async def async_get_config_entry_diagnostics(
# a single physical unit exposes more than one logical Device. See
# registry/identity.py.
identity = coordinator._identity
def _sub_unit_diag(su) -> dict:
# One pass over coordinator.bound for both fields below (count and
# the distinct hrefs), and one redaction of this unit's canonical
# view -- `model` reads modelNum off the already-redacted `resources`
# rather than redacting /information/vs/0 a second time. modelNum
# itself never matches redact.py's substring rules, so which side of
# redact_resources it's read from doesn't change the value.
matching = [b for b in coordinator.bound if b.sub_unit == su]
res = redact_resources(coordinator.canonical_resources(su))
return {
"kind": su.kind,
"key": su.key,
"seed_path": "/" + "/".join(su.seed_path),
"bound_entity_count": len(matching),
"hrefs": sorted({b.href for b in matching}),
"model": res.get('/information/vs/0', {}).get('x.com.samsung.da.modelNum', ''),
# Keyed by this unit's *canonical* hrefs, not the real ones
# it answers on -- '/mode/vs/0' rather than '/mode/vs/1' or
# '/<uuid>/mode/vs/0'. That's the form the registry and every
# capability are written against, so a sibling's block can be
# read (or pasted into the skill's standalone-discovery
# recipe) exactly like the master's `resources` above,
# instead of having to be de-indexed by hand first.
"resources": res,
}
return {
"device_type": coordinator.device_type_name or "unknown",
"one_ui_version": coordinator.one_ui_version,
@@ -49,7 +77,67 @@ async def async_get_config_entry_diagnostics(
"resources": redact_resources(identity.raw),
} if identity is not None else None,
"unbound_hrefs": sorted(coordinator._unbound_hrefs),
"resources": redact_resources(coordinator.last_resources),
# This unit's own resources, and only this unit's -- what the
# module docstring and the adding-device-support skill have always
# described it as ("the parsed /device/0 snapshot"). On a composite
# device (issue #177) `last_resources` is the union across every
# live unit keyed by real hrefs, so reporting it raw here would mix
# a sibling's /mode/vs/1 in with the master's /mode/vs/0 under no
# attribution at all. Each sibling reports its own resources in its
# own `sub_units` entry below instead. For a device with no sub-units
# -- almost every device -- this is byte-identical to `last_resources`.
"resources": redact_resources(coordinator.canonical_resources(MAIN)),
# Sibling indoor units discovered on this connection (issue #177) --
# per-unit kind/key/seed path plus what actually bound to it, so a
# report shows whether a composite device's sub-unit was found at
# all and what it resolved to. subdeviceIdList (the UUID a prefixed
# unit's key comes from) is deliberately NOT redacted here even
# though the field matches redact.py's 'deviceid' substring rule
# elsewhere in `resources` above -- it's an appliance-internal
# pairing id, not account data, and reporting the key is what makes
# this block actionable.
"sub_units": [_sub_unit_diag(su) for su in coordinator.sub_units],
# Candidates that answered their seed but that discover_partitioned's
# entity-level liveness gate rejected -- an unused SmartThings slot
# (HJcom's /device/2) that still answers a same-shaped batch, not a
# real second unit. Reported alongside sub_units above so a report
# shows what was found *and* why it didn't become an entity, not
# just silence where a third climate card might otherwise be
# expected.
"sub_units_skipped": [
{
"kind": skip.sub_unit.kind,
"key": skip.sub_unit.key,
"seed_path": "/" + "/".join(skip.sub_unit.seed_path),
"hrefs": list(skip.hrefs),
# The reps the liveness gate actually judged, canonicalized
# like the materialized units above. These are the one thing
# a reader needs to second-guess a skip ("is my second unit
# really absent, or did the gate get it wrong?"), and they
# exist nowhere else in this dump: a rejected candidate is
# never polled again and never enters the state cache, so
# `resources` above cannot contain them by construction.
"resources": redact_resources({
canon: rep
for href, rep in coordinator._skipped_sub_unit_resources.items()
if (canon := skip.sub_unit.to_canonical(href)) is not None
}),
}
for skip in coordinator._skipped_sub_units
],
# What each enumeration probe returned ({} vs a batch), keyed by the
# seed href attempted -- lets a report distinguish "checked, nothing
# there" from "never checked", the same posture the speculative
# /device/1 //device/2 probe this replaced used to document directly
# in identity.py before it moved to registry/subunits.py.
"sub_unit_probes": dict(sorted(coordinator._sub_unit_probes.items())),
# /multidevice/vs/0's rep ({} when the board doesn't answer it).
# Reported on its own rather than inside `resources` because it is
# metadata about the connection rather than state of any one unit --
# and because nothing polls it after discovery, so it would go stale
# in there. Its numofsubdevice count is what independently
# corroborates the sub_units/sub_units_skipped split above.
"multidevice": redact_resources(coordinator._multidevice),
"integration_version": integration.version,
"smartthings_local_version": stl_version,
"observe_mode": coordinator.observe_mode,
+22 -3
View File
@@ -33,12 +33,19 @@ def _is_included(bound: BoundEntity, coordinator: 'LocalThingsCoordinator') -> b
a field is genuinely never populated on unsupported hardware opts into
stricter gating with its own is_stub_rep-based exists_fn (see
common.ENERGY_METER, issue #127) -- this default stays permissive.
`bound.href` is already the *actual* href (issue #177 -- see
BoundEntity/SubUnit), so the direct cache lookup below is correct as-is;
`exists_fn` gets `bound`'s own sub-unit's *canonical* view instead of the
raw snapshot, same rule as everywhere else a whole-resources-dict scan
happens (coordinator.canonical_resources) -- this is a free function, not
an LocalThingsEntity method, so it can't use self._resources.
"""
rep = coordinator.last_resources.get(bound.href)
if rep is None:
return False
if bound.desc.exists_fn is not None:
return bound.desc.exists_fn(rep, coordinator.last_resources)
return bound.desc.exists_fn(rep, coordinator.canonical_resources(bound.sub_unit))
if bound.desc.field:
if not rep or is_stub_rep(rep):
return True
@@ -122,9 +129,21 @@ class LocalThingsEntity(CoordinatorEntity[LocalThingsCoordinator]):
"""
tk = self._bound.desc.translation_key
if callable(tk):
return tk(self.coordinator.last_resources)
return tk(self._resources)
return tk if tk is not None else self._bound.desc.key
@property
def _resources(self) -> dict:
"""This entity's own sub-unit's canonical resources view (issue
#177) -- see coordinator.canonical_resources. Every platform
property that needs the *whole* resources dict, as opposed to one
href via `coordinator.resource(href)`, must read through this
instead of `coordinator.last_resources`, or a sibling unit's own
actual hrefs would leak into (or be missing from) this entity's
view. For MAIN (every device with no sub-units) this is exactly
`coordinator.last_resources`."""
return self.coordinator.canonical_resources(self._bound.sub_unit)
@property
def device_info(self) -> DeviceInfo:
return self.coordinator.device_info
return self.coordinator.device_info_for(self._bound.sub_unit)
+3 -3
View File
@@ -134,7 +134,7 @@ class LocalThingsRangeHoodFan(LocalThingsEntity, FanEntity):
def _power_payload(self, enabled: bool) -> tuple[str, bool, str]:
"""Target whichever power resource this hood actually exposes."""
resources = self.coordinator.last_resources
resources = self._resources
target = POWER_HREF if POWER_HREF in resources else POWER_VS_HREF
return 'power', enabled, target
@@ -237,7 +237,7 @@ class LocalThingsAirPurifierFan(LocalThingsEntity, FanEntity):
Writing a hardcoded href here would silently no-op on a board that
only reports the other one, even though is_on already falls back
correctly."""
resources = self.coordinator.last_resources
resources = self._resources
target = POWER_VS_HREF if POWER_VS_HREF in resources else POWER_HREF
return 'power', enabled, target
@@ -344,7 +344,7 @@ class LocalThingsAirflowFan(LocalThingsEntity, FanEntity):
resources -- disagreeing until the next poll refreshes the other
one (the same optimistic-apply lag coordinator.py's own comments
warn about)."""
resources = self.coordinator.last_resources
resources = self._resources
target = POWER_HREF if POWER_HREF in resources else POWER_VS_HREF
return 'power', enabled, target
@@ -4,19 +4,44 @@ from __future__ import annotations
from typing import Any
from .discovery import BoundEntity
from .subunits import SubUnit, canonical_view
def _key(b: BoundEntity) -> str:
return f"{b.key_override or b.desc.key}{b.instance}"
# b.sub_unit.key_prefix is '' for MAIN, so a device with no sub-units
# (every device this integration shipped before issue #177) gets a
# byte-identical key to before -- a hard regression guard, not a nicety
# (see test_unique_ids.py and every golden file under tests/fixtures/golden/).
return f"{b.sub_unit.key_prefix}{b.key_override or b.desc.key}{b.instance}"
def flatten(bound: list[BoundEntity], resources: dict) -> dict[str, Any]:
"""Map bound entities to their current scalar values."""
"""Map bound entities to their current scalar values.
`exists_fn(rep, resources)` receives that entity's own sub-unit's
*canonical* resources view (see subunits.canonical_view), not the raw
actual-href snapshot -- an exists_fn that scans the whole resources dict
for a sibling href (e.g. is_legacy_board) must judge each unit on its
own resources, not see another unit's hrefs bleed in under the same
canonical key. Views are built once per distinct sub-unit per call, not
once per entity -- O(sub-units), not O(bound entities).
"""
out: dict[str, Any] = {}
# SubUnit is a frozen dataclass (hashable, equal by value), so it can key
# `views` directly -- no need to re-derive an identity for it out of
# (kind, key) first.
all_units = list(dict.fromkeys(b.sub_unit for b in bound))
views: dict[SubUnit, dict] = {}
for b in bound:
rep = resources.get(b.href) or {}
if b.desc.exists_fn is not None and not b.desc.exists_fn(rep, resources):
continue
if b.desc.exists_fn is not None:
view = views.get(b.sub_unit)
if view is None:
view = canonical_view(b.sub_unit, resources, all_units)
views[b.sub_unit] = view
if not b.desc.exists_fn(rep, view):
continue
if b.desc.rep_fn is not None:
out[_key(b)] = b.desc.rep_fn(rep)
elif b.desc.field:
@@ -957,6 +957,15 @@ _AC_IGNORED = [
# Undocumented single int (runningMode: 0 on every dump seen), no
# supported-values list to interpret it against -- 'don't guess'.
'/runn/vs/0',
# 2-in-1/multi-indoor-unit systems (issue #177, HJcom's
# ARTIK051_DONGLE_FAC_18K): x.com.samsung.da.numofsubdevice, a plain
# corroborating count of indoor units on this connection. Confirmed
# read-only (a write attempt returned CoAP 4.00). Absent from
# /device/0's batch entirely -- registry.subunits.enumerate_sub_units
# fetches it with its own RETRIEVE and folds it into the resources dict
# for diagnostics, which is why it needs an entry here rather than
# surfacing as an unbound-href gap on every board that has it.
'/multidevice/vs/0',
]
# Built as bare no-entity caps; folded into the AC registry (not global).
@@ -18,6 +18,7 @@ from typing import Callable, Iterable, Optional
from .capability import Capability
from .entities import SamsungEntityDescription
from .subunits import MAIN, SubUnit
@dataclass
@@ -28,6 +29,11 @@ class BoundEntity:
instance: str = ''
key_override: Optional[str] = None
instance_name: Optional[str] = None
# Which logical indoor unit (issue #177) this entity belongs to. `href`
# above is always the *actual*, on-the-wire href for that unit -- MAIN's
# to_actual is the identity transform, so every device with no sub-units
# behaves exactly as before this field existed.
sub_unit: SubUnit = MAIN
def _snake_to_title(s: str) -> str:
@@ -57,13 +63,20 @@ def instance_suffix(href: str) -> str:
def _bind(cap: Capability, href: str, inst: str, inst_name: Optional[str],
key_prefix: Optional[str] = None) -> list[BoundEntity]:
key_prefix: Optional[str] = None,
sub_unit: SubUnit = MAIN) -> list[BoundEntity]:
"""Build one BoundEntity per entity on `cap`, sharing the instance/
key-prefix/instance-name computed once by the caller."""
key-prefix/instance-name computed once by the caller.
`href` here is the *canonical* href discover() is iterating over;
`sub_unit.to_actual` maps it to the real, on-the-wire href the entity
actually reads/writes (identity for MAIN, so single-unit devices are
unaffected -- see subunits.py)."""
return [
BoundEntity(href=href, capability=cap, desc=desc, instance=inst,
BoundEntity(href=sub_unit.to_actual(href), capability=cap, desc=desc,
instance=inst,
key_override=f'{key_prefix}_{desc.key}' if key_prefix else None,
instance_name=inst_name)
instance_name=inst_name, sub_unit=sub_unit)
for desc in cap.entities
]
@@ -74,6 +87,7 @@ def discover(
pattern_caps: Iterable[Capability] = (),
log: Optional[Callable[[str], None]] = None,
tier_log: Optional[Callable[[str, str], None]] = None,
sub_unit: SubUnit = MAIN,
) -> list[BoundEntity]:
"""`tier_log(href, poll_tier)` fires for every href a capability actually
matches, even a no-entity "coverage-only" capability (see COVERAGE lists
@@ -81,7 +95,18 @@ def discover(
Callers that need a href's poll cadence (the coordinator's hot/warm
sub-poll and OBSERVE-attempt lists) must use this, not `bound` -- a
coverage-only capability's `poll_tier` would otherwise be silently
dropped since it never appears in `bound`."""
dropped since it never appears in `bound`.
`resources` is always keyed by *canonical* hrefs -- for a sub-unit
(issue #177) that means its own canonical view (see
subunits.canonical_view), the same shape as a plain single-unit device's
resources dict, so registry lookups/rt_filter/match_fn/instance_suffix
all behave identically regardless of which unit is being discovered.
`sub_unit` only affects the *href* stamped onto each BoundEntity (via
`_bind`, see above) and the href `log`/`tier_log` report -- both the
real, subscribable/pollable path, not the canonical one the registry is
keyed on.
"""
out: list[BoundEntity] = []
for href, rep in resources.items():
@@ -97,10 +122,10 @@ def discover(
if cap.match_fn is not None and not cap.match_fn(rep, resources):
continue
inst = instance_suffix(href)
out.extend(_bind(cap, href, inst, _instance_name(cap, rep)))
out.extend(_bind(cap, href, inst, _instance_name(cap, rep), sub_unit=sub_unit))
matched = True
if tier_log is not None:
tier_log(href, cap.poll_tier)
tier_log(sub_unit.to_actual(href), cap.poll_tier)
if matched:
continue
@@ -117,13 +142,14 @@ def discover(
# Auto-derive key prefix from href segments (skip digits and 'vs')
src = href[len(cap.href_prefix):] if (cap.strip_prefix_in_key and cap.href_prefix) else href
segs = [s for s in src.strip('/').split('/') if s and not s.isdigit() and s != 'vs']
out.extend(_bind(cap, href, inst, _instance_name(cap, rep), '_'.join(segs)))
out.extend(_bind(cap, href, inst, _instance_name(cap, rep), '_'.join(segs),
sub_unit=sub_unit))
matched = True
if tier_log is not None:
tier_log(href, cap.poll_tier)
tier_log(sub_unit.to_actual(href), cap.poll_tier)
break
if not matched and not caps and log is not None:
log(href)
log(sub_unit.to_actual(href))
return out
@@ -6,20 +6,6 @@ from typing import Optional
import cbor2
from .batch import parse_device0_batch
# Speculative /device/<n> siblings to probe alongside the coordinator's own
# /device/0 seed poll (issue #177). Confirmed against a real dump: /oic/res's
# baseline-Interface response only lists resources with the discoverable
# policy bit set, and /device/0's whole x.com.samsung.da.* tree is registered
# without it -- so a second logical Device's Collection, if one exists, would
# be just as invisible to /oic/res as /device/0 is. A direct GET is the only
# way left to check, and it's a plain RETRIEVE (non-mutating, tolerated-404
# already the norm throughout this module) -- not the kind of guess the
# write-contract 'don't guess' rule is about. Bounded to a couple of indices;
# widen only if a real Composite Device ever turns out to need more.
_SPECULATIVE_DEVICE_INDICES = (1, 2)
@dataclass(frozen=True)
class DeviceIdentity:
@@ -56,24 +42,6 @@ def _get_links(sess, path) -> list:
return []
def _get_device_batch(sess, index: int) -> dict[str, dict]:
"""GET /device/<index> and parse it the same way the coordinator parses
/device/0 -- a Samsung Collection RETRIEVE returns
[devcol-rep, {href, rep}, {href, rep}, ...], not a bare Property map or
Link array. Missing/malformed responses fall through to {} (via
parse_device0_batch on an empty/non-list body), same tolerated-absence
posture as _get/_get_links above."""
try:
code, pl = sess.get(['device', str(index)], timeout=10.0)
if code == 0x45 and pl:
body = cbor2.loads(pl)
if isinstance(body, list):
return parse_device0_batch(body)
except Exception:
pass
return {}
def _device_types(d: dict) -> tuple[str, ...]:
"""/oic/d's `rt` -- the device's own OCF device-type declaration.
@@ -104,14 +72,13 @@ def read_identity(sess, serial: Optional[str]) -> DeviceIdentity:
# Relevant for the OCF "Composite Device" model (issue #177: a single
# physical unit -- one IP, one /oic/p -- exposing more than one logical
# Device, each as its own Collection resource, same rt shape as our own
# /device/0). Nothing consumes this yet; captured so a report from a
# multi-unit device shows us whether its firmware actually implements
# that model before any code assumes it does.
# /device/0). This is what registry.subunits.enumerate_sub_units reads
# to find a board's `/device/<n>` siblings (Pattern A -- HJcom's
# ARTIK051_DONGLE_FAC_18K) -- that probing, plus the /device/1 and
# /device/2 speculative fallback it used to run right here on every
# _connect_session (including every reconnect), moved to that module so
# it only runs once, at first discovery, instead of on every reconnect.
res = _get_links(sess, ['oic', 'res'])
extra_devices = {
f'/device/{n}': _get_device_batch(sess, n)
for n in _SPECULATIVE_DEVICE_INDICES
}
return DeviceIdentity(
manufacturer=p.get('mnmn') or 'Samsung',
model=p.get('mnmo') or '',
@@ -121,8 +88,5 @@ def read_identity(sess, serial: Optional[str]) -> DeviceIdentity:
# Kept whole rather than field-by-field: these resources are outside
# the /device/0 dump diagnostics already captures, and we don't yet
# know which of their fields will turn out to identify a device type.
# /device/1 and /device/2 are always present here (empty {} when the
# device didn't answer) so a diagnostics reader can tell "checked,
# nothing there" apart from "never checked".
raw={'/oic/p': p, '/oic/d': d, '/oic/res': res, **extra_devices},
raw={'/oic/p': p, '/oic/d': d, '/oic/res': res},
)
@@ -0,0 +1,492 @@
"""Sub-unit ("composite device") support for one physical connection exposing
more than one logical indoor unit -- issue #177.
Two reporters, two different board families, two genuinely different
mechanisms for exposing a second indoor unit over one IP / one DTLS session
(see DESIGN-177.md section 1 for the full evidence trail; the two diagnostics
dumps this was built against are HJcom's and jhkwon19's -- they each filed
one of the two reports this module unifies):
Pattern A -- indexed siblings (`ARTIK051_DONGLE_FAC_18K`, HJcom's board).
`/oic/res` lists three complete parallel resource sets whose trailing path
segment is the index (`/mode/vs/0`, `/mode/vs/1`, `/mode/vs/2`, ... on both
OCF-standard and vendor hrefs), and `/device/0`'s batch carries only the
index-0 hrefs -- the sibling units are reachable only via their own
`/device/<n>` collection.
Pattern B -- UUID-prefixed tree (`TP2X_FAC_BORA_21K`, jhkwon19's board).
`/oic/res` hides the whole appliance tree; `/device/0`'s batch instead
carries `x.com.samsung.da.subdeviceIdList` on `/subdevices/vs/0`, and that
same UUID appears as a literal href prefix in `/oic/res`
(`/<uuid>/file/list/vs/0`, ...). `GET /<uuid>/device/0` returns the second
unit's own Collection batch, confirmed live by the reporter to carry a
different model/serial than the master (`TP2X_FAC_BORA_RAC_21K`, the
wall-mounted unit, vs. the master's `TP2X_FAC_BORA_21K`, the floor unit).
Both are "the same thing wearing different clothes": a logical unit is a
seed collection path to poll, plus an href transform between the canonical
href the registry knows (`/mode/vs/0`) and the actual on-the-wire href. The
detection signals don't overlap on either captured board (HJcom's has no
`/subdevices/vs/0` at all; jhkwon19's has no `/device/1`), so no
disambiguation logic is needed -- `enumerate_sub_units` checks both and
materializes any candidate whose seed answers with a non-empty batch.
A non-empty seed batch is necessary but not sufficient for the *candidate*
to actually be a live second unit, though: HJcom's own board also has a
`/device/2` -- a third, unused SmartThings slot -- that answers with the
exact same 14-href shape as the real `/device/1` sibling, populated with
three constant/echoed/shape-only reps (a region code identical to every
other unit's, an /information rep echoing the *same* model string as unit
1, and a /temperatures items[] entry with an id/description but no
current/desired/minimum/maximum reading) and nothing resembling live
climate state. Gating on *resource* shape/hrefs turned out to be the wrong
layer -- it would need per-family domain knowledge (which hrefs mean "in
use" for a washer's second drum, a fridge's second compartment, ...) baked
into a registry field before any of those families could use this module
at all. `discover_partitioned` instead gates at the *entity* layer, after
discovery+flattening: a candidate is only kept if it produced at least one
*primary* (no `entity_category`) bound entity whose flattened value isn't
`None` -- e.g. HJcom's /device/2 does flatten to an `alarm_code` value, but
that entity is diagnostic-category and derived from an empty /alarms/vs/2,
so it doesn't count. This reuses the same primary/config/diagnostic
taxonomy every registry already declares (see the adding-device-support
skill's entity-taxonomy section) instead of adding a second, parallel
domain-knowledge mechanism.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from typing import Callable, Optional
import cbor2
from .batch import parse_device0_batch
_INDEXED_HREF_RE = re.compile(r'^/device/(\d+)$')
# Speculative /device/<n> siblings probed when /oic/res doesn't reveal a
# second logical Device's Collection on this board (moved here from
# identity.py, issue #177 -- see enumerate_sub_units' docstring for why: the
# old read_identity fired these two extra RETRIEVEs on *every* _connect_session,
# including every reconnect, for information enumeration only needs once).
# Same bound as before: a plain, tolerated-404 RETRIEVE, not the kind of
# guess the write-contract 'don't guess' rule is about. Widen only if a real
# board ever turns out to need more than two siblings.
_SPECULATIVE_DEVICE_INDICES = (1, 2)
@dataclass(frozen=True)
class SubUnit:
"""One logical indoor unit reachable over a single physical connection.
`kind='main'` is the unit this config entry actually connects to and
always exists (see MAIN below) -- its `to_actual`/`to_canonical` are the
identity transform, so every existing single-unit device keeps behaving
exactly as it did before this module existed. `'indexed'`/`'prefixed'`
are Pattern A/B above; `key` is the trailing index string ('1', '2', ...)
or the full subdevice UUID, and `seed_path` is the Collection href (as
path segments) whose batch response enumerates/refreshes that unit.
"""
kind: str # 'main' | 'indexed' | 'prefixed'
key: str # '' | '1' | '6c2dff6d-ee5c-dad1-6a5e-000000000001'
seed_path: tuple[str, ...]
def to_actual(self, canonical: str) -> str:
"""Canonical registry href (e.g. '/mode/vs/0') -> the real,
on-the-wire href for this unit."""
if self.kind == 'indexed':
head, sep, tail = canonical.rpartition('/')
# Only the index-0 trailing segment is ours to rewrite --
# deliberately not a "replace any trailing digit" rule, which
# would misread a genuine multi-instance resource (the fridge's
# pattern-cap hrefs, e.g. '/door/vs/1') as a sub-unit's. No
# registry declares a non-zero trailing index today and no
# fixture in the corpus contains one (verified across the whole
# corpus), so the strict rule costs nothing.
if tail == '0':
return f'{head}{sep}{self.key}'
return canonical
if self.kind == 'prefixed':
return f'/{self.key}{canonical}'
return canonical
def to_canonical(self, actual: str) -> Optional[str]:
"""Inverse of to_actual, or None when `actual` isn't this unit's."""
if self.kind == 'indexed':
head, sep, tail = actual.rpartition('/')
if tail == self.key:
return f'{head}{sep}0'
return None
if self.kind == 'prefixed':
prefix = f'/{self.key}'
if actual.startswith(prefix + '/'):
return actual[len(prefix):]
return None
return actual
def owns(self, actual: str) -> bool:
"""True if `actual` belongs to this unit's namespace. MAIN never
"owns" anything by this definition -- it gets whatever's left after
every other sub-unit's hrefs are excluded (see canonical_view)."""
if self.kind == 'main':
return False
return self.to_canonical(actual) is not None
@property
def key_prefix(self) -> str:
"""Prefix that guarantees a unique entity key/unique_id (see
adapter._key). '' for MAIN -- the master unit's flattened state
keys must stay byte-identical to every device this integration
shipped before issue #177, so no golden file changes. The full
subdevice UUID is used verbatim (non-alphanumerics stripped, not
truncated or replaced with an ordinal) because it's device-reported
and stable across reconnects/restarts, unlike an ordinal assigned by
enumeration order -- and it never appears in a user-visible string
(see DESIGN-177.md section 6): HA derives the visible entity_id from
the device name + entity name, not from unique_id.
"""
if self.kind == 'indexed':
return f'unit{self.key}_'
if self.kind == 'prefixed':
slug = re.sub(r'[^a-zA-Z0-9]', '', self.key)
return f'sub_{slug}_'
return ''
MAIN = SubUnit(kind='main', key='', seed_path=('device', '0'))
def canonical_view(
sub_unit: SubUnit, resources: dict[str, dict], sub_units: list['SubUnit'],
) -> dict[str, dict]:
"""Rewrite `resources` (real, on-the-wire hrefs) into `sub_unit`'s own
canonical namespace -- what discover()/exists_fn/rep_fn/is_legacy_board
and friends are written against.
For MAIN this is the snapshot *minus* every href owned by one of the
other units in `sub_units` -- otherwise a sibling's own `/mode/vs/1`
would leak into the master's view under the same canonical key
('/mode/vs/0') that the master's actual `/mode/vs/0` also maps to,
silently mixing two units' state together. For an indexed/prefixed unit
it's the reverse: only the hrefs that unit owns, rewritten back through
`to_canonical`.
`sub_units` may or may not include MAIN itself -- MAIN.owns() is always
False, so including it is harmless.
"""
if sub_unit.kind == 'main':
owned_elsewhere = {
href for href in resources
if any(su.owns(href) for su in sub_units)
}
return {h: r for h, r in resources.items() if h not in owned_elsewhere}
return {
canon: resources[actual]
for actual in resources
if (canon := sub_unit.to_canonical(actual)) is not None
}
def normalize_seed_batch(sub_unit: SubUnit, batch: dict[str, dict]) -> dict[str, dict]:
"""Real, on-the-wire hrefs from one sub-unit's seed-collection batch,
normalized so every href actually carries this unit's prefix/index.
Indexed units need no change -- the device echoes the real `/x/<n>`
href in its own `/device/<n>` batch (confirmed against HJcom's dump).
A prefixed unit's batch entries may or may not already carry the
`/<id>` prefix (unconfirmed which -- jhkwon19's board was never probed
live before the sub-unit id was known), so it's added when missing.
"""
if sub_unit.kind != 'prefixed':
return batch
prefix = f'/{sub_unit.key}'
return {
(href if href.startswith(prefix + '/') else f'{prefix}{href}'): rep
for href, rep in batch.items()
}
def _iter_oic_res_hrefs(oic_res):
"""Flatten /oic/res's raw shape into a flat iterable of link dicts.
Both captured dumps group links by `di` (`[{'di': ..., 'links': [...]}]`
-- see identity.py's read_identity/_get_links), so that's the shape
handled here. Tolerant of a flat link-list too (nothing in the OCF spec
rules it out, and _get_links' own posture already treats any list-shaped
body as possible) and of anything else by yielding nothing.
"""
for entry in (oic_res or []):
if not isinstance(entry, dict):
continue
if 'links' in entry:
for link in entry.get('links') or []:
if isinstance(link, dict):
yield link
elif 'href' in entry:
yield entry
def _seed_href(path_segs: tuple[str, ...]) -> str:
"""('device', '1') -> '/device/1' -- the leading-slash href form
`probe_log` and diagnostics report, built from the path-segment form
`sess.get` takes."""
return '/' + '/'.join(path_segs)
def _get_raw(sess, path_segs: tuple[str, ...]):
"""GET `path_segs` and CBOR-decode the payload, or None on any
missing/malformed response (a 4.04, a timeout, an empty payload) --
shared tolerated-absence posture for both callers below, which differ
only in which body shape they accept."""
try:
code, pl = sess.get(list(path_segs), timeout=10.0)
if code == 0x45 and pl:
return cbor2.loads(pl)
except Exception:
pass
return None
def _get_batch(sess, path_segs: tuple[str, ...]) -> dict[str, dict]:
"""GET a Samsung Collection resource and parse it the same way
/device/0 itself is parsed (parse_device0_batch): a [devcol-rep,
{href, rep}, ...] CBOR list, not a bare Property map."""
body = _get_raw(sess, path_segs)
return parse_device0_batch(body) if isinstance(body, list) else {}
def _get_property(sess, path_segs: tuple[str, ...]) -> dict:
"""GET a plain OCF Property-map resource (a bare dict, not a Collection
batch). Used for `/multidevice/vs/0` (issue #177 follow-up): listed in
`/oic/res` on HJcom's board but absent from `/device/0`'s batch, so it
needs its own RETRIEVE, and it answers a single Property map, not a
[devcol-rep, ...] list."""
body = _get_raw(sess, path_segs)
return body if isinstance(body, dict) else {}
def enumerate_sub_units(
sess,
resources: dict[str, dict],
oic_res_links,
probe_log: Optional[Callable[[str, bool], None]] = None,
) -> tuple[list['SubUnit'], dict[str, dict]]:
"""Discover every sibling indoor unit reachable over `sess`'s connection.
Runs once, at first discovery, in an executor, under the coordinator's
session lock -- every GET here is a plain RETRIEVE (the write-contract
'don't guess' rule doesn't apply to reading an extra resource to find
out whether it's there). Returns the *candidate* units and the resources
already fetched while probing them (already normalized to real hrefs),
so the coordinator's first discovery poll doesn't need to re-poll them.
`probe_log(seed_href, found)` fires for every seed attempted, whether or
not it answered -- so diagnostics (see diagnostics.py's sub_unit_probes)
can tell "checked, nothing there" apart from "never checked", the same
posture the speculative-probe code this replaces used to document in
identity.py.
Every candidate whose seed answers with a non-empty batch is returned
here -- this function has no way to tell a real sibling from an unused
SmartThings slot that merely answers the same shape (HJcom's
`/device/2`); that requires discovering+flattening the candidate's own
entities first, which is `discover_partitioned`'s job, not this one's.
See this module's docstring.
"""
units: list[SubUnit] = []
fetched: dict[str, dict] = {}
def _probed(seed_href: str, batch: dict) -> None:
if probe_log is not None:
probe_log(seed_href, bool(batch))
# --- Pattern B: UUID-prefixed tree (TP2X_FAC_BORA_21K) ------------------
raw_ids = (resources.get('/subdevices/vs/0') or {}).get(
'x.com.samsung.da.subdeviceIdList')
# Tolerate anything but a list of strings -- this field is redaction-prone
# (it matches the 'deviceid' substring rule in redact.py) and the existing
# airconditioner_fac_bora fixture carries the literal string
# '**REDACTED**'/'REDACTED' there. That must yield zero sub-units, not a
# crash -- issue #177 is additive, it must never break an already-working
# single-climate-entity device.
ids = raw_ids if isinstance(raw_ids, list) else []
for sub_id in sorted(i for i in ids if isinstance(i, str) and i):
seed = (sub_id, 'device', '0')
batch = _get_batch(sess, seed)
_probed(_seed_href(seed), batch)
if not batch:
continue
unit = SubUnit(kind='prefixed', key=sub_id, seed_path=seed)
fetched.update(normalize_seed_batch(unit, batch))
units.append(unit)
# --- Pattern A: indexed siblings (ARTIK051_DONGLE_FAC_18K) --------------
indices = sorted({
int(m.group(1))
for link in _iter_oic_res_hrefs(oic_res_links)
for m in [_INDEXED_HREF_RE.match(link.get('href', ''))]
if m and int(m.group(1)) >= 1
})
if not indices:
# A board that hides its whole tree from /oic/res (Pattern B's
# jhkwon19 board does this too, but it has no /device/<n> to find
# regardless) gives us nothing to enumerate from -- fall back to the
# bounded speculative probe this replaces from identity.py.
indices = list(_SPECULATIVE_DEVICE_INDICES)
for n in indices:
seed = ('device', str(n))
batch = _get_batch(sess, seed)
_probed(_seed_href(seed), batch)
if not batch:
continue
unit = SubUnit(kind='indexed', key=str(n), seed_path=seed)
fetched.update(batch) # already real /x/<n> hrefs, no normalization needed
units.append(unit)
# /multidevice/vs/0 (issue #177 follow-up): HJcom's board lists it in
# /oic/res but it never appears in /device/0's batch, so it needs its
# own RETRIEVE. It's a plain corroborating count
# (x.com.samsung.da.numofsubdevice), confirmed read-only (a write
# attempt returned CoAP 4.00) -- captured for diagnostics only, folded
# into the merged resources dict like any other href (see
# airconditioner._AC_IGNORED, which is what keeps it from surfacing as
# an unbound-href gap). NOT a gate: discover_partitioned's entity-level
# liveness check decides materialization correctly without it, and only
# this one board family is known to expose it at all. Whether it agrees
# with the number of units actually materialized is the coordinator's
# call to log (it owns the logger; this module doesn't), not this
# function's.
multidevice_seed = ('multidevice', 'vs', '0')
multidevice = _get_property(sess, multidevice_seed)
_probed(_seed_href(multidevice_seed), multidevice)
if multidevice:
fetched['/multidevice/vs/0'] = multidevice
return units, fetched
@dataclass(frozen=True)
class SkippedSubUnit:
"""A candidate `enumerate_sub_units` found whose seed answered, but that
`discover_partitioned`'s entity-level liveness gate rejected -- an
unused SmartThings slot (HJcom's `/device/2`), not a real second unit.
Kept around (rather than silently dropped) so a caller can log/report
what was skipped and why."""
sub_unit: SubUnit
hrefs: tuple[str, ...]
def _has_live_primary_entity(bound, state: dict) -> bool:
"""True if flattening `bound` (one candidate sub-unit's BoundEntity
list) produced at least one non-`None` value for a *primary* entity --
`entity_category` unset, HA's own "the user acts on or watches this"
tier (see the adding-device-support skill's entity-taxonomy section).
This is the materialization gate itself (see this module's docstring):
HJcom's `/device/2` does flatten to one non-`None` value
(`alarm_code`), but that entity is `diagnostic`-category and derived
from an empty `/alarms/vs/2` -- a config/diagnostic entity reading
"something" proves nothing about whether a physical unit is actually
installed there, so it's deliberately excluded from this check.
"""
from .adapter import _key # see discover_partitioned's deferred-import note
return any(
not b.desc.entity_category and state.get(_key(b)) is not None
for b in bound
)
def discover_partitioned(
resources: dict[str, dict],
sub_units: list['SubUnit'],
resolve_registry: Callable[[dict], object],
fallback_capabilities: dict,
log: Optional[Callable[[str], None]] = None,
tier_log: Optional[Callable[[str, str], None]] = None,
):
"""Bind every href in `resources` (the merged, real-href snapshot -- main
plus every enumerated sub-unit's seed) to entities, partitioned by which
unit owns it.
Main pass runs over hrefs owned by no sub-unit -- otherwise every
`/mode/vs/1` would land in `unbound_hrefs` too (nothing in the main
device's registry claims that literal href) and raise a spurious
coverage-gap repair. Then one pass per *candidate* sub-unit over its own
canonical view, resolving that unit's own device type from its own
`/information/vs/0` when it reports one (e.g. jhkwon19's wall unit
reports `TP2X_FAC_BORA_RAC_21K` -> the 'RAC' board token ->
airconditioner), falling back to the master's registry otherwise --
every AC family shares the same resource surface, and a sibling that
fails to answer its own identity resource is still the same appliance
type as the unit this config entry was set up against.
Each candidate is discovered and flattened *twice*: once silently to
evaluate `_has_live_primary_entity` (this module's materialization
gate -- see its docstring and this module's own), and, only if that
passes, a second time with `log`/`tier_log` wired so its coverage gaps
and poll tiers actually count. A candidate that fails the gate
contributes nothing at all -- no bound entities, no unbound-href
report, no hot/warm href -- as if it had never answered its seed.
Discovering an unused slot's small, fixed resource set twice at
first-discovery time only is a non-issue; getting a phantom sub-unit
silently counted into unbound_hrefs or hot/warm tiers is not.
Returns `(bound, device_type_name, materialized, skipped)`:
- `bound`: the concatenated BoundEntity list (main + every materialized
sub-unit).
- `device_type_name`: the *master's* resolved device type (used for
logging/device naming; each sub-unit's own resolved type only affects
which capabilities bind its hrefs, not this).
- `materialized`: the subset of `sub_units` that passed the gate, in the
same order -- what the caller should keep as its live sub-unit roster
going forward (poll seeds, canonical_resources, device_info_for, ...).
- `skipped`: `SkippedSubUnit` entries for every candidate that didn't.
"""
# Deferred import: discovery.py imports SubUnit/MAIN from this module at
# module scope, so importing discover() back here at module scope would
# be a circular import. By the time this function actually runs both
# modules are fully loaded. adapter.py imports discovery.py, so the same
# applies to flatten()/_key().
from .adapter import flatten
from .discovery import discover
# Same computation canonical_view does for MAIN (snapshot minus every
# other sub-unit's owned hrefs) -- reuse it rather than re-deriving
# owned_elsewhere here too.
main_view = canonical_view(MAIN, resources, sub_units)
reg = resolve_registry(main_view)
caps, pats = (
(reg.capabilities, reg.pattern_capabilities) if reg is not None
else (fallback_capabilities, [])
)
# MAIN is never gated -- the config entry's own physical connection
# always materializes regardless of what its entities' values are.
bound = discover(main_view, caps, pats, log=log, tier_log=tier_log, sub_unit=MAIN)
device_type_name = reg.name if reg is not None else None
materialized: list[SubUnit] = []
skipped: list[SkippedSubUnit] = []
for su in sub_units:
view = canonical_view(su, resources, sub_units)
su_reg = resolve_registry(view) or reg
su_caps, su_pats = (
(su_reg.capabilities, su_reg.pattern_capabilities) if su_reg is not None
else (fallback_capabilities, [])
)
probe_bound = discover(view, su_caps, su_pats, sub_unit=su)
probe_state = flatten(probe_bound, resources)
if _has_live_primary_entity(probe_bound, probe_state):
materialized.append(su)
bound = bound + discover(
view, su_caps, su_pats, log=log, tier_log=tier_log, sub_unit=su,
)
else:
skipped.append(SkippedSubUnit(
sub_unit=su,
hrefs=tuple(sorted({b.href for b in probe_bound})),
))
return bound, device_type_name, materialized, skipped
+3 -1
View File
@@ -104,7 +104,9 @@ class LocalThingsSelect(LocalThingsEntity, SelectEntity):
# list decoded from a sibling resource. There is no static
# fallback: when that resource isn't populated the callable
# returns [] and the entity's exists_fn suppresses it entirely.
return list(desc.options(self.coordinator.last_resources) or [])
# This entity's own sub-unit's canonical view (issue #177), not
# the raw actual-href snapshot -- see LocalThingsEntity._resources.
return list(desc.options(self._resources) or [])
if desc.options_field:
rep = self.coordinator.last_resources.get(self._bound.href) or {}
return list(rep.get(desc.options_field) or [])
+83
View File
@@ -16,6 +16,89 @@ def _load_device(name: str) -> dict[str, dict]:
return _resources_from_dump(data)
def _load_device_full(name: str):
"""Like _load_device, but also returns the optional `oic_res`/`seeds`
keys a sub-unit-capable fixture (issue #177) may carry alongside
`device0` -- see the two `airconditioner_*` fixtures with a
`seeds_note` field. `oic_res`/`seeds` default to `[]`/`{}` for every
other fixture, so this is safe to call on any fixture in the corpus.
Returns `(resources, oic_res, seeds)` where `seeds` is
`{seed_href: raw_batch_list}` -- the same [devcol-rep, {href, rep}, ...]
shape a real /device/<n> or /<id>/device/0 RETRIEVE returns, ready to
hand to a FakeCoapSession.
A fixture's optional `probes` map (plain Property-map resources that
belong to no batch, e.g. the hand-read /multidevice/vs/0 in the
ARTIK051_DONGLE_FAC_18K fixture) is folded into `seeds` here, since
FakeCoapSession answers both shapes off the same href key.
"""
data = json.loads((FIXTURES / f'{name}_device.json').read_text())
resources = _resources_from_dump(data)
seeds = {**data.get('seeds', {}), **data.get('probes', {})}
return resources, data.get('oic_res', []), seeds
class FakeCoapSession:
"""Minimal stand-in for smartthings_local's DtlsCoapSession, backed by a
fixture's `seeds` map (raw device0-batch-shaped lists keyed by seed
href -- plus any `probes` entries, which are plain Property maps rather
than batch lists; both are just CBOR bodies at this layer, and the two
readers in registry.subunits already type-check what they get back).
Enough surface for registry.subunits.enumerate_sub_units and
LocalThingsCoordinator's blocking sub-unit polls to run against fixture
data without a live device -- same idea as test_identity.py's
FakeSession, but keyed by href string (post path-join) rather than a
path tuple, since callers here pass a `seed_path` tuple straight
through.
"""
def __init__(self, seeds: dict[str, list] | None = None):
self.seeds = seeds or {}
def get(self, path, timeout=None):
href = '/' + '/'.join(path)
body = self.seeds.get(href)
if body is None:
return 0x84, b'' # 4.04 not found -- tolerated absence
import cbor2
return 0x45, cbor2.dumps(body)
def pace(self):
pass
def _discover_full(resources: dict[str, dict], oic_res, seeds: dict[str, list]):
"""Run the *whole* sub-unit-aware discovery pipeline against fixture
data, HA-free -- mirrors exactly what LocalThingsCoordinator does across
_enumerate_sub_units_blocking + _run_discovery (issue #177), so a test
exercising this exercises the real code path, not a re-implementation of
it. See the adding-device-support skill's section 2 for the plain
(non-sub-unit) equivalent this extends.
Returns `(bound, materialized, skipped, full_resources, device_type_name)`:
- `bound`: every BoundEntity, main + every materialized sub-unit.
- `materialized`/`skipped`: SubUnit / SkippedSubUnit lists straight from
discover_partitioned.
- `full_resources`: `resources` merged with every candidate's seed data
(actual hrefs) -- what a coordinator's cache would hold.
- `device_type_name`: the master's resolved registry name.
"""
from custom_components.localthings.registry.by_type import resolve
from custom_components.localthings.registry.registry import CAPABILITIES
from custom_components.localthings.registry.subunits import (
discover_partitioned, enumerate_sub_units,
)
sess = FakeCoapSession(seeds)
candidates, extra = enumerate_sub_units(sess, resources, oic_res)
full_resources = {**resources, **extra}
bound, device_type_name, materialized, skipped = discover_partitioned(
full_resources, candidates, resolve, CAPABILITIES,
)
return bound, materialized, skipped, full_resources, device_type_name
def _load_resources(ip: str) -> dict[str, dict]:
"""Legacy IP-based loader — maps known IPs to named fixtures."""
_ip_to_name = {
File diff suppressed because it is too large Load Diff
+809
View File
@@ -0,0 +1,809 @@
{
"device0": [
{
"rt": [
"x.com.samsung.devcol",
"oic.wk.col"
],
"if": [
"oic.if.baseline",
"oic.if.ll",
"oic.if.b"
]
},
{
"href": "/personality/presence/vs/0",
"rep": {
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "",
"x.com.samsung.da.deviceId": "REDACTED",
"x.com.samsung.da.value": ""
}
]
}
},
{
"href": "/realtimenotiforclient/vs/0",
"rep": {
"x.com.samsung.da.timeforshortnoti": "0",
"x.com.samsung.da.longnotisubscription": "true",
"x.com.samsung.da.periodicnotisubscription": "true"
}
},
{
"href": "/filter/airdustfilter/vs/0",
"rep": {
"x.com.samsung.da.filterUsage": "0",
"x.com.samsung.da.filterUsageResolution": "1",
"x.com.samsung.da.filterDesiredUsage": "112",
"x.com.samsung.da.filterStatus": "normal",
"x.com.samsung.da.filterCapacity": "112",
"x.com.samsung.da.filterCapacityUnit": "Hour",
"x.com.samsung.da.filterResetType": [
"washable"
],
"x.com.samsung.da.supportedFilterDesiredUsage": [
"112",
"224",
"336",
"448"
]
}
},
{
"href": "/temperature/control/vs/0",
"rep": {
"x.com.samsung.da.increment": "1"
}
},
{
"href": "/mode/convenient/vs/0",
"rep": {
"x.com.samsung.da.modes": "Off",
"x.com.samsung.da.supportedModes": [
"Off",
"Sleep",
"Quiet",
"Speed"
]
}
},
{
"href": "/option/autoclean/vs/0",
"rep": {
"x.com.samsung.da.status": "Stop",
"x.com.samsung.da.settingStatus": "On",
"x.com.samsung.da.progress": "0",
"x.com.samsung.da.supportedStatus": [
"Start",
"SpeedClean",
"QuietClean",
"Stop"
],
"x.com.samsung.da.supportedSettingStatus": [
"On",
"SpeedClean",
"QuietClean",
"Off"
]
}
},
{
"href": "/wind/strength/vs/0",
"rep": {
"x.com.samsung.da.modes": "2",
"x.com.samsung.da.supportedModes": [
"0",
"2",
"3",
"4"
],
"x.com.samsung.da.modesName": [
"Auto",
"Mid",
"High",
"Turbo"
]
}
},
{
"href": "/wind/direction/vs/0",
"rep": {
"x.com.samsung.da.modes": "NotSupported",
"x.com.samsung.da.supportedModes": [
"NotSupported"
]
}
},
{
"href": "/subdevices/vs/0",
"rep": {
"x.com.samsung.da.subdeviceIdList": [
"6c2dff6d-ee5c-dad1-6a5e-000000000001"
]
}
},
{
"href": "/alarms/vs/0",
"rep": {
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Alarm",
"x.com.samsung.da.alarmType": "Device",
"x.com.samsung.da.code": "ErrorCode_OFF",
"x.com.samsung.da.triggeredTime": "2026-07-29T04:31:31"
},
{
"x.com.samsung.da.id": "2",
"x.com.samsung.da.description": "Alarm",
"x.com.samsung.da.alarmType": "Device",
"x.com.samsung.da.code": "AC_V_0002_OFF",
"x.com.samsung.da.triggeredTime": "2026-07-29T04:31:31"
}
]
}
},
{
"href": "/temperatures/vs/0",
"rep": {
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Temperature",
"x.com.samsung.da.desired": "24.0",
"x.com.samsung.da.current": "33.0",
"x.com.samsung.da.maximum": "30",
"x.com.samsung.da.minimum": "18",
"x.com.samsung.da.increment": "1.0",
"x.com.samsung.da.unit": "Celsius"
}
]
}
},
{
"href": "/temperature/current/0",
"rep": {
"range": [
18.0,
30.0
],
"units": "C",
"temperature": 33.0
}
},
{
"href": "/temperature/desired/0",
"rep": {
"range": [
18.0,
30.0
],
"units": "C",
"temperature": 24.0
}
},
{
"href": "/diagnosis/vs/0",
"rep": {}
},
{
"href": "/energy/consumption/vs/0",
"rep": {
"x.com.samsung.da.cumulativeConsumption": "100.000000",
"x.com.samsung.da.instantaneousPower": "65278.000000",
"x.com.samsung.da.usageThreshold": "0.000000",
"x.com.samsung.da.cumulativePower": "544088",
"x.com.samsung.da.cumulativeUnit": "Wh",
"x.com.samsung.da.instantaneousPowerUnit": "W",
"x.com.samsung.da.cumulativePowerType": "individual"
}
},
{
"href": "/energy/consumption/0",
"rep": {
"energy": 100.0,
"power": 65278.0
}
},
{
"href": "/mode/vs/0",
"rep": {
"x.com.samsung.da.supportedModes": [
"AIComfort",
"Cool",
"Dry",
"Wind"
],
"x.com.samsung.da.modes": [
"Wind"
],
"x.com.samsung.da.options": [
"Operation_Family",
"Blooming_0",
"OnTimer_0",
"OffTimer_0",
"Sleep_16",
"ArtificialWorking_Off",
"ComfortAICooling_Off",
"AiTempChanged_Off",
"AiTemp_270",
"welcomecare_Off",
"Panel_Close",
"Weather_Off",
"Volume_100",
"StopAutoClean_Idle",
"DiagnosisAI_Off",
"Display_Off",
"ProgressDiagnosisAI_1",
"ResultDiagnosisAI_Normal",
"Service_Off",
"SmartCoolClean_Off",
"ProgressSmartClean_0",
"OutDoorVentil_Off",
"FreezeAlarmSetting_Off",
"DesiredFreezeAlarm_240",
"OptionCode_529",
"ExtendOptionCode_16975",
"RacInfo_First",
"RacInfo_None_Second",
"ModelInfo_16K_BORA_VENT2",
"UpdateAllow_NotAllowed",
"EnergySaveIcon_Off",
"DurationOn_0",
"welcomecareElapsedTime_0",
"welcomecareThresholdTemp_0",
"welcomecareStartDate_0000",
"welcomecareEndDate_0000",
"welcomecareSeason_None"
]
}
},
{
"href": "/power/vs/0",
"rep": {
"x.com.samsung.da.power": "Off"
}
},
{
"href": "/power/0",
"rep": {
"value": false
}
},
{
"href": "/information/vs/0",
"rep": {
"x.com.samsung.da.modelNum": "TP2X_FAC_BORA_21K|10233041|600001110015110006000C1200830000",
"x.com.samsung.da.description": "TP2X_FAC_BORA_21K",
"x.com.samsung.da.serialNum": "REDACTED",
"x.com.samsung.da.otnDUID": "REDACTED",
"x.com.samsung.da.diagProtocolType": "WIFI_HTTPS",
"x.com.samsung.da.diagLogType": [
"errCode",
"dump"
],
"x.com.samsung.da.diagDumpType": "file",
"x.com.samsung.da.diagEndPoint": "SSM",
"x.com.samsung.da.diagMnid": "0AJT",
"x.com.samsung.da.diagSetupid": "000",
"x.com.samsung.da.diagMinVersion": "1.0",
"x.com.samsung.da.serialNumOption": "REDACTED",
"x.com.samsung.da.items": [
{
"x.com.samsung.da.id": "0",
"x.com.samsung.da.description": "Version",
"x.com.samsung.da.type": "Software",
"x.com.samsung.da.number": "02337A260424",
"x.com.samsung.da.newVersionAvailable": "0"
},
{
"x.com.samsung.da.id": "1",
"x.com.samsung.da.description": "Version",
"x.com.samsung.da.type": "Firmware",
"x.com.samsung.da.number": "2102240021022200",
"x.com.samsung.da.newVersionAvailable": "0"
},
{
"x.com.samsung.da.id": "2",
"x.com.samsung.da.description": "Version",
"x.com.samsung.da.type": "Outdoor",
"x.com.samsung.da.number": "2103300110000300"
}
]
}
},
{
"href": "/file/information/vs/0",
"rep": {
"x.com.samsung.timeoffset": "+09:00",
"x.com.samsung.supprtedtype": 1
}
},
{
"href": "/configuration/vs/0",
"rep": {
"x.com.samsung.da.region": "3017000000",
"x.com.samsung.da.airconOptionList": [
"HOMECARE_WIZARD_V2",
"ENERGY_2.0",
"AI_2.0",
"DeviceTypeMaster",
"SingleCommand_1"
]
}
},
{
"href": "/humidity/0",
"rep": {
"humidity": 0
}
},
{
"href": "/humidity/vs/0",
"rep": {
"x.com.samsung.da.humidity": "0.000000",
"x.com.samsung.da.fivepercentHumidity": "58"
}
},
{
"href": "/drlc/0",
"rep": {
"DRLevel": 2,
"start": "2026-07-29T04:12:57Z",
"duration": 20,
"override": false
}
},
{
"href": "/drlc/vs/0",
"rep": {
"x.com.samsung.da.drlcLevel": "2",
"x.com.samsung.da.duration": "20:30:00",
"x.com.samsung.da.drlcStartTime": "2026-07-29T04:12:57Z",
"x.com.samsung.da.override": "Off"
}
},
{
"href": "/runn/vs/0",
"rep": {
"x.com.samsung.da.runningMode": 0
}
},
{
"href": "/availablecontrolsets/vs/0",
"rep": {
"x.com.samsung.da.sets": "000000B4012C0000404B04000000",
"x.com.samsung.da.id": "FAC",
"x.com.samsung.da.version": "1.0"
}
},
{
"href": "/keepnormalstate/vs/0",
"rep": {
"x.com.samsung.da.keepnormal": 5
}
},
{
"href": "/otninformation/vs/0",
"rep": {
"x.com.samsung.da.target": "",
"x.com.samsung.da.newVersionAvailable": "false",
"x.com.samsung.da.newVersionNo": "00000000",
"x.com.samsung.da.currentVersionInfo": "00000000",
"otnStatus": "None",
"flashingProgress": "",
"otnTarget": "main",
"otnCompleteDate": "noHistory",
"otnList": [
{
"type": "WIFI",
"modelId": "AFA-KR-TP2-21-AF9X00",
"versions": [
"10260424"
],
"visVersion": "260424"
},
{
"type": "Micom",
"modelId": "04511023304110232941",
"versions": [
"21022400",
"21022200"
],
"visVersion": "210224"
},
{
"type": "Micom",
"modelId": "04511022974110229941",
"versions": [
"21033001",
"10000300"
],
"visVersion": "210330"
},
{
"type": "Micom",
"modelId": "045110230741FFFFFFFF",
"versions": [
"22050300",
"FFFFFFFF"
],
"visVersion": "220503"
}
]
}
},
{
"href": "/timezone/vs/0",
"rep": {
"timezoneid": "Asia/Seoul",
"offset": "+09:00",
"DST": "OFF"
}
}
],
"oic_res": [
{
"di": "**REDACTED**",
"links": [
{
"href": "/oic/sec/doxm",
"rt": [
"oic.r.doxm"
],
"if": [
"oic.if.baseline"
],
"p": {
"bm": 1,
"sec": true,
"port": 49154,
"x.org.iotivity.tls": 0
}
},
{
"href": "/oic/sec/pstat",
"rt": [
"oic.r.pstat"
],
"if": [
"oic.if.baseline"
],
"p": {
"bm": 1,
"sec": true,
"port": 49154,
"x.org.iotivity.tls": 0
}
},
{
"href": "/oic/d",
"rt": [
"oic.wk.d",
"oic.d.airconditioner"
],
"if": [
"oic.if.baseline",
"oic.if.r"
],
"p": {
"bm": 1,
"sec": false,
"x.org.iotivity.tcp": 0
}
},
{
"href": "/oic/p",
"rt": [
"oic.wk.p"
],
"if": [
"oic.if.baseline",
"oic.if.r"
],
"p": {
"bm": 1,
"sec": false,
"x.org.iotivity.tcp": 0
}
},
{
"href": "/hass/state/vs/0",
"rt": [
"x.com.samsung.da.hass.state"
],
"if": [
"oic.if.baseline",
"oic.if.a"
],
"p": {
"bm": 3,
"sec": false,
"x.org.iotivity.tcp": 0
}
},
{
"href": "/hass/command/vs/0",
"rt": [
"x.com.samsung.da.hass.command"
],
"if": [
"oic.if.baseline",
"oic.if.a"
],
"p": {
"bm": 3,
"sec": false,
"x.org.iotivity.tcp": 0
}
},
{
"href": "/file/transfer/chunk/vs/0",
"rt": [
"x.com.samsung.file.chunk"
],
"if": [
"oic.if.baseline",
"oic.if.a"
],
"p": {
"bm": 1,
"sec": false,
"x.org.iotivity.tcp": 0
}
},
{
"href": "/file/list/vs/0",
"rt": [
"x.com.samsung.file.list"
],
"if": [
"oic.if.baseline",
"oic.if.s"
],
"p": {
"bm": 1,
"sec": false,
"x.org.iotivity.tcp": 0
}
},
{
"href": "/file/transfer/vs/0",
"rt": [
"x.com.samsung.file.transfer"
],
"if": [
"oic.if.baseline",
"oic.if.a"
],
"p": {
"bm": 3,
"sec": false,
"x.org.iotivity.tcp": 0
}
},
{
"href": "/6c2dff6d-ee5c-dad1-6a5e-000000000001/file/list/vs/0",
"rt": [
"x.com.samsung.file.list"
],
"if": [
"oic.if.baseline",
"oic.if.s"
],
"p": {
"bm": 1,
"sec": false,
"x.org.iotivity.tcp": 0
}
},
{
"href": "/6c2dff6d-ee5c-dad1-6a5e-000000000001/file/transfer/vs/0",
"rt": [
"x.com.samsung.file.transfer"
],
"if": [
"oic.if.baseline",
"oic.if.a"
],
"p": {
"bm": 3,
"sec": false,
"x.org.iotivity.tcp": 0
}
},
{
"href": "/EasySetupResURI",
"rt": [
"oic.r.easysetup"
],
"if": [
"oic.if.baseline",
"oic.if.ll",
"oic.if.b"
],
"p": {
"bm": 1,
"sec": true,
"port": 49154,
"x.org.iotivity.tls": 0
}
},
{
"href": "/WiFiConfResURI",
"rt": [
"oic.wk.wifi"
],
"if": [
"oic.if.baseline"
],
"p": {
"bm": 1,
"sec": true,
"port": 49154,
"x.org.iotivity.tls": 0
}
},
{
"href": "/CoapCloudConfResURI",
"rt": [
"oic.wk.cloudserver"
],
"if": [
"oic.if.baseline"
],
"p": {
"bm": 1,
"sec": true,
"port": 49154,
"x.org.iotivity.tls": 0
}
},
{
"href": "/DevConfResURI",
"rt": [
"oic.wk.devconf"
],
"if": [
"oic.if.baseline"
],
"p": {
"bm": 1,
"sec": true,
"port": 49154,
"x.org.iotivity.tls": 0
}
},
{
"href": "/sec/provisioninginfo",
"rt": [
"x.com.samsung.provisioninginfo"
],
"if": [
"oic.if.baseline",
"oic.if.a"
],
"p": {
"bm": 1,
"sec": false,
"x.org.iotivity.tcp": 0
}
},
{
"href": "/sec/accesspointlist",
"rt": [
"x.com.samsung.accesspointlist"
],
"if": [
"oic.if.baseline",
"oic.if.s"
],
"p": {
"bm": 1,
"sec": false,
"x.org.iotivity.tcp": 0
}
}
]
}
],
"seeds": {
"/6c2dff6d-ee5c-dad1-6a5e-000000000001/device/0": [
{
"rt": [
"x.com.samsung.devcol",
"oic.wk.col"
],
"if": [
"oic.if.baseline",
"oic.if.ll",
"oic.if.b"
]
},
{
"href": "/information/vs/0",
"rep": {
"x.com.samsung.da.modelNum": "TP2X_FAC_BORA_RAC_21K|10233041|600001110015110006000C1200830000",
"x.com.samsung.da.description": "TP2X_FAC_BORA_RAC_21K",
"x.com.samsung.da.serialNum": "TEST-SUBUNIT-SERIAL-0000"
}
},
{
"href": "/power/vs/0",
"rep": {
"x.com.samsung.da.power": "On"
}
},
{
"href": "/mode/vs/0",
"rep": {
"x.com.samsung.da.supportedModes": [
"Cool",
"Dry",
"Wind",
"Auto"
],
"x.com.samsung.da.modes": [
"Cool"
],
"x.com.samsung.da.options": []
}
},
{
"href": "/temperature/current/0",
"rep": {
"range": [
18.0,
30.0
],
"units": "C",
"temperature": 26.0
}
},
{
"href": "/temperature/desired/0",
"rep": {
"range": [
18.0,
30.0
],
"units": "C",
"temperature": 24.0
}
},
{
"href": "/wind/strength/vs/0",
"rep": {
"x.com.samsung.da.modes": "0",
"x.com.samsung.da.supportedModes": [
"0",
"2",
"3",
"4"
],
"x.com.samsung.da.modesName": [
"Auto",
"Mid",
"High",
"Turbo"
]
}
},
{
"href": "/wind/direction/vs/0",
"rep": {
"x.com.samsung.da.modes": [
"Fix"
],
"x.com.samsung.da.supportedModes": [
"Fix",
"All"
]
}
}
]
},
"seeds_note": "device0 and oic_res are real, captured from jhkwon19's diagnostics dump for the TP2X_FAC_BORA_21K board (issue #177, the same physical unit as tests/fixtures/airconditioner_fac_bora_device.json -- that fixture's redacted x.com.samsung.da.subdeviceIdList string is a deliberate, separate regression case and is NOT changed here). subdeviceIdList above is restored to the real ['6c2dff6d-ee5c-dad1-6a5e-000000000001'] reported in the issue thread (redacted in the raw capture the same way serialNum/otnDUID/deviceId are -- it matches redact.py's 'deviceid' substring rule, not because it's actually account data). The seed's /information/vs/0 rep is REAL, verbatim from the reporter's live debug-panel read of /6c2dff6d-.../information/vs/0 (serial replaced with a placeholder) -- modelNum TP2X_FAC_BORA_RAC_21K confirms this is the wall-mounted room unit, distinct from the master's floor-unit TP2X_FAC_BORA_21K. Every other href in this seed batch is CONSTRUCTED (never read from this unit) so the sub-unit's climate entity has a resource surface to bind -- per DESIGN-177.md section 4, retrieval for this pattern is batch-only, and confirming '/6c2dff6d-.../device/0' actually returns a full batch (as opposed to just answering the information probe the reporter tried by hand) is listed as a follow-up, not something this fixture can attest to."
}
@@ -0,0 +1,27 @@
{
"state_keys": [
"alarm_code",
"auto_clean_legacy",
"beep",
"clean_level",
"climate",
"current_temperature_c",
"diagnosis_status",
"dust",
"energy_kwh",
"fine_dust",
"good_sleep",
"humidity",
"odor",
"power_energy_kwh",
"power_watts",
"super_fine_dust",
"unit1_alarm_code",
"unit1_auto_clean_legacy",
"unit1_beep",
"unit1_climate",
"unit1_current_temperature_c",
"unit1_good_sleep",
"unit1_humidity"
]
}
+22
View File
@@ -0,0 +1,22 @@
{
"state_keys": [
"air_filter_status",
"air_filter_threshold",
"air_filter_usage",
"air_filter_usage_hours",
"alarm_code",
"auto_clean",
"beep",
"climate",
"current_temperature_c",
"diagnosis_status",
"energy_kwh",
"firmware_update",
"humidity",
"power_energy_kwh",
"power_watts",
"sub_6c2dff6dee5cdad16a5e000000000001_climate",
"sub_6c2dff6dee5cdad16a5e000000000001_current_temperature_c",
"tropical_night_mode"
]
}
+6
View File
@@ -20,6 +20,12 @@ class _FakeCoordinator:
def resource(self, href):
return self.last_resources.get(href, {})
def canonical_resources(self, sub_unit):
# Every bound entity in this test uses the default MAIN
# sub-unit, so the canonical view is just the raw snapshot
# (issue #177 -- see LocalThingsEntity._resources).
return self.last_resources
async def async_send_command(self, bound, payload):
self.commands.append((bound, payload))
+6
View File
@@ -29,6 +29,12 @@ class _FakeCoordinator:
def resource(self, href):
return self.last_resources.get(href, {})
def canonical_resources(self, sub_unit):
# Every bound entity in this test uses the default MAIN
# sub-unit, so the canonical view is just the raw snapshot
# (issue #177 -- see LocalThingsEntity._resources).
return self.last_resources
async def async_send_command(self, bound, payload):
self.commands.append((bound, payload))
@@ -47,6 +47,12 @@ class _FakeCoordinator:
def resource(self, href):
return self.last_resources.get(href, {})
def canonical_resources(self, sub_unit):
# Every bound entity in this test uses the default MAIN
# sub-unit, so the canonical view is just the raw snapshot
# (issue #177 -- see LocalThingsEntity._resources).
return self.last_resources
async def async_send_command(self, bound, payload):
self.commands.append((bound, payload))
@@ -34,6 +34,12 @@ class _FakeCoordinator:
def resource(self, href):
return self.last_resources.get(href, {})
def canonical_resources(self, sub_unit):
# Every bound entity in this test uses the default MAIN
# sub-unit, so the canonical view is just the raw snapshot
# (issue #177 -- see LocalThingsEntity._resources).
return self.last_resources
async def async_send_command(self, bound, payload):
self.commands.append((bound, payload))
+5
View File
@@ -106,6 +106,11 @@ def test_fac_bora_wind_strength_codes_fit_the_standard_scale():
def resource(self, href):
return self.last_resources.get(href, {})
def canonical_resources(self, sub_unit):
# This test's entity uses the default MAIN sub-unit, so the
# canonical view is just the raw snapshot (issue #177).
return self.last_resources
resources = _load_device('airconditioner_fac_bora')
info = resources['/information/vs/0']
reg = by_type.for_device_by_model(
+103
View File
@@ -0,0 +1,103 @@
"""Tests that a sub-unit's LocalThingsClimate entity (issue #177) reads its
*own* power/mode/temperature -- not the master's, and not some mix of the
two -- and that the legacy-board test (is_legacy_board/_legacy_airflow) is
evaluated per unit rather than once globally.
Uses HJcom's ARTIK051_DONGLE_FAC_18K fixture deliberately: it's a legacy
`/airflow/vs/<n>` board (no `/wind/*` at all) on *both* the master and its
materialized sibling, which is exactly the shape climate.py's own comments
warn is easy to get wrong if the canonical view leaks between units.
"""
from __future__ import annotations
import pytest
from homeassistant.components.climate import HVACMode
from homeassistant.core import HomeAssistant
from custom_components.localthings.climate import LocalThingsClimate
from custom_components.localthings.registry.entities import ClimateDesc
from tests.test_subdevice_discovery import _coordinator, _discover
def _climate_entities(coordinator):
"""{sub_unit_key_or_None: LocalThingsClimate}, None standing for MAIN."""
from custom_components.localthings.registry.subunits import MAIN
out = {}
for b in coordinator.bound:
if isinstance(b.desc, ClimateDesc):
key = None if b.sub_unit == MAIN else b.sub_unit.key
out[key] = LocalThingsClimate(coordinator, b)
return out
@pytest.fixture
async def climates(hass: HomeAssistant):
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_artik051_dongle_fac_18k')
return _climate_entities(coordinator)
async def test_sub_unit_climate_reads_its_own_mode_and_power(climates):
"""Master reports 'Auto'; the bedroom unit (/device/1) reports 'Cool' --
confirmed distinct in the real captured fixture. If the sub-unit
entity's _rep() weren't translating through its own sub_unit, it would
read the master's /mode/vs/0 instead and report the master's mode."""
main, unit1 = climates[None], climates['1']
assert main.hvac_mode == HVACMode.AUTO
assert unit1.hvac_mode == HVACMode.COOL
async def test_sub_unit_climate_reads_its_own_temperature(climates):
"""Master: current 25.0 / desired 26.0. Unit 1: current 27.0 / desired
28.0 -- distinct values in the real fixture, so a href mix-up here
would show up as a wrong number, not just a wrong mode string."""
main, unit1 = climates[None], climates['1']
assert main.current_temperature == 25.0
assert main.target_temperature == 26.0
assert unit1.current_temperature == 27.0
assert unit1.target_temperature == 28.0
async def test_sub_unit_climate_reads_its_own_power_state(climates):
"""Both units happen to report power On in this fixture -- this at
least confirms _is_on() reads the *unit's own* /power/vs/<n>, not a
hardcoded /power/vs/0, by checking the entity resolves without falling
back to OFF (which _is_on() would do if it silently read an absent
href instead of the sub-unit's actual one)."""
main, unit1 = climates[None], climates['1']
assert main.hvac_mode != HVACMode.OFF
assert unit1.hvac_mode != HVACMode.OFF
async def test_legacy_board_test_is_evaluated_per_unit(climates):
"""HJcom's board has no /wind/* resources at all on *either* unit --
is_legacy_board(self._resources) must independently evaluate True for
the master's own canonical view and for the sub-unit's own canonical
view. If is_legacy_board were fed the raw, unpartitioned snapshot (or
if canonical_view leaked one unit's hrefs into the other's), this
wouldn't distinguish "this unit is legacy" from "some unit on this
connection is legacy" -- and a future board with one legacy + one
modern unit sharing a connection would silently read the wrong fan/
swing channel on one side."""
main, unit1 = climates[None], climates['1']
assert main._legacy_airflow() != {}
assert unit1._legacy_airflow() != {}
# Both resolve to *some* fan mode via the legacy path rather than the
# /wind/strength/vs/0 channel (absent on this board) -- confirms
# is_legacy_board(self._resources) actually gated fan_mode's branch,
# not just that _legacy_airflow() itself returned something.
assert main.fan_mode is not None
assert unit1.fan_mode is not None
async def test_sub_unit_climate_writes_are_scoped_to_its_own_bound_entity(climates):
"""A sanity check that the two entities are backed by genuinely
different BoundEntity objects (different hrefs), which is what makes
per-unit reads/writes possible at all -- see async_send_command's
translation test in test_coordinator_send_command.py for the write
side of this."""
main, unit1 = climates[None], climates['1']
assert main._bound.href == '/mode/vs/0'
assert unit1._bound.href == '/mode/vs/1'
assert main._bound is not unit1._bound
+86
View File
@@ -25,7 +25,10 @@ from custom_components.localthings.const import (
)
from custom_components.localthings.coordinator import LocalThingsCoordinator
from custom_components.localthings.registry.capabilities import laundry
from custom_components.localthings.registry.capabilities.airconditioner import _climate_write
from custom_components.localthings.registry.discovery import BoundEntity
from custom_components.localthings.registry.entities import ClimateDesc
from custom_components.localthings.registry.subunits import SubUnit
ENTRY_DATA = {
CONF_HOST: '10.0.0.199',
@@ -92,3 +95,86 @@ async def test_options_write_optimistic_cache_keeps_sibling_tokens(coordinator)
assert cached['x.com.samsung.da.options'] == [
'DeviceType_0167', 'Course_1D', 'GMT_04',
]
# ---------------------------------------------------------------------------
# Sub-unit write translation (issue #177): a composite-entity write_fn (here
# airconditioner._climate_write) returns *canonical* path_segs
# (['power', 'vs', '0']) -- async_send_command must translate that through
# bound_entity.sub_unit.to_actual before POSTing, applying the optimistic
# value, and starting the settle guard, or a sub-unit's climate card would
# write to (and read confirmation from) the master's resource instead of
# its own.
# ---------------------------------------------------------------------------
def _climate_bound(href: str, sub_unit: SubUnit) -> BoundEntity:
desc = ClimateDesc(key='climate', translation_key='airconditioner',
write_fn=_climate_write)
return BoundEntity(href=href, capability=None, desc=desc, sub_unit=sub_unit)
async def test_indexed_sub_unit_write_posts_to_translated_path(coordinator) -> None:
"""A power write from the bedroom unit's (indexed '1') climate entity
must POST to /power/vs/1, not the master's /power/vs/0."""
unit1 = SubUnit(kind='indexed', key='1', seed_path=('device', '1'))
bound = _climate_bound('/mode/vs/1', unit1)
await coordinator.async_send_command(bound, ('power', True))
posted_path, posted_bytes = coordinator._session.post_calls[0]
assert posted_path == ['power', 'vs', '1']
assert cbor2.loads(posted_bytes) == {'x.com.samsung.da.power': 'On'}
async def test_indexed_sub_unit_write_applies_optimistic_value_to_translated_href(
coordinator,
) -> None:
unit1 = SubUnit(kind='indexed', key='1', seed_path=('device', '1'))
bound = _climate_bound('/mode/vs/1', unit1)
await coordinator.async_send_command(bound, ('power', True))
# The optimistic apply + settle guard must land on /power/vs/1 -- the
# translated href a real device confirms this write against -- not on
# /mode/vs/1 (bound_entity.href) or /power/vs/0 (the master's resource).
assert coordinator._cache.get('/power/vs/1') == {'x.com.samsung.da.power': 'On'}
assert coordinator._cache.get('/power/vs/0') is None
# The settle guard is armed on that same translated href: a stale poll
# reporting the pre-write value must be dropped, not allowed to revert
# the optimistic 'On' (issue #27's regression, translated to a
# sub-unit's own resource).
applied = coordinator._observe.apply(
'/power/vs/1', {'x.com.samsung.da.power': 'Off'}, source='poll',
)
assert applied is False
assert coordinator._cache.get('/power/vs/1') == {'x.com.samsung.da.power': 'On'}
async def test_prefixed_sub_unit_write_posts_to_translated_path(coordinator) -> None:
"""A power write from a UUID-prefixed unit's climate entity must POST
to /<uuid>/power/vs/0, not the bare canonical href."""
sub_id = '6c2dff6d-ee5c-dad1-6a5e-000000000001'
unit = SubUnit(kind='prefixed', key=sub_id, seed_path=(sub_id, 'device', '0'))
bound = _climate_bound(f'/{sub_id}/mode/vs/0', unit)
await coordinator.async_send_command(bound, ('power', True))
posted_path, posted_bytes = coordinator._session.post_calls[0]
assert posted_path == [sub_id, 'power', 'vs', '0']
assert cbor2.loads(posted_bytes) == {'x.com.samsung.da.power': 'On'}
assert coordinator._cache.get(f'/{sub_id}/power/vs/0') == {
'x.com.samsung.da.power': 'On',
}
async def test_main_climate_write_unaffected_by_sub_unit_translation(coordinator) -> None:
"""MAIN's to_actual is the identity transform -- a device with no
sub-units must keep posting to the exact same path as before this
translation step existed."""
from custom_components.localthings.registry.subunits import MAIN
bound = _climate_bound('/mode/vs/0', MAIN)
await coordinator.async_send_command(bound, ('power', True))
posted_path, _ = coordinator._session.post_calls[0]
assert posted_path == ['power', 'vs', '0']
+107
View File
@@ -0,0 +1,107 @@
"""Sanity check for diagnostics.py's issue #177 additions (sub_units,
sub_units_skipped, sub_unit_probes) against a real composite-device
discovery run -- makes sure the new blocks are actually reachable/shaped
right, not just that the coordinator's own attributes look correct in
isolation (test_subdevice_discovery.py covers that)."""
from __future__ import annotations
from homeassistant.core import HomeAssistant
from custom_components.localthings.const import DOMAIN
from custom_components.localthings.diagnostics import (
async_get_config_entry_diagnostics,
)
from tests.test_subdevice_discovery import _coordinator, _discover
async def test_diagnostics_reports_materialized_and_skipped_sub_units(
hass: HomeAssistant, enable_custom_integrations,
) -> None:
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_artik051_dongle_fac_18k')
hass.data.setdefault(DOMAIN, {})[coordinator._entry.entry_id] = coordinator
diag = await async_get_config_entry_diagnostics(hass, coordinator._entry)
sub_unit_keys = {su['key'] for su in diag['sub_units']}
skipped_keys = {su['key'] for su in diag['sub_units_skipped']}
assert sub_unit_keys == {'1'}
assert skipped_keys == {'2'}
assert diag['sub_units'][0]['bound_entity_count'] > 0
assert diag['sub_units_skipped'][0]['hrefs']
assert '/multidevice/vs/0' in diag['sub_unit_probes']
# The reporter's hand-read value, carried in the fixture's `probes` map
# (it belongs to no batch -- see that fixture's seeds_note). Two real
# units, master + bedroom, independently corroborating the gate's
# decision to skip /device/2. Reported on its own, never as a resource.
assert diag['multidevice'] == {'x.com.samsung.da.numofsubdevice': '2'}
assert '/multidevice/vs/0' not in diag['resources']
async def test_top_level_resources_is_the_master_unit_only(
hass: HomeAssistant, enable_custom_integrations,
) -> None:
"""`resources` reports this unit's own hrefs and nothing else.
It used to be `last_resources` raw -- the union across every live unit,
keyed by real hrefs -- so a sibling's /mode/vs/1 sat in it alongside the
master's /mode/vs/0 with no attribution, contradicting both the module
docstring and the adding-device-support skill ("the parsed /device/0
snapshot"). Each sibling carries its own canonicalized block instead.
"""
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_artik051_dongle_fac_18k')
hass.data.setdefault(DOMAIN, {})[coordinator._entry.entry_id] = coordinator
diag = await async_get_config_entry_diagnostics(hass, coordinator._entry)
assert not [h for h in diag['resources'] if h.endswith('/1')]
assert '/mode/vs/0' in diag['resources']
# The sibling's own block carries its state, canonicalized -- '/mode/vs/0',
# not the '/mode/vs/1' it actually answers on.
unit1 = diag['sub_units'][0]
assert '/mode/vs/0' in unit1['resources']
assert not [h for h in unit1['resources'] if h.endswith('/1')]
assert unit1['resources']['/power/vs/0']['x.com.samsung.da.power'] == 'On'
async def test_rejected_candidate_reps_reach_diagnostics_but_not_the_cache(
hass: HomeAssistant, enable_custom_integrations,
) -> None:
"""A gate-rejected slot is never polled again, so anything applied to the
state cache for it would sit frozen at its first-discovery value while
looking as live as every other href. It's kept out of the cache entirely
and reported only under its own sub_units_skipped entry -- which is also
the only place a reader can go to second-guess the gate."""
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_artik051_dongle_fac_18k')
hass.data.setdefault(DOMAIN, {})[coordinator._entry.entry_id] = coordinator
assert not [h for h in coordinator.last_resources if h.endswith('/2')]
diag = await async_get_config_entry_diagnostics(hass, coordinator._entry)
skipped = diag['sub_units_skipped'][0]
# Present, canonicalized, and visibly the empty state the gate rejected.
assert skipped['resources']['/power/vs/0'] == {}
assert skipped['resources']['/mode/vs/0'] == {}
assert skipped['resources']['/information/vs/0']
async def test_diagnostics_reports_prefixed_sub_unit(
hass: HomeAssistant, enable_custom_integrations,
) -> None:
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_fac_bora_2in1')
hass.data.setdefault(DOMAIN, {})[coordinator._entry.entry_id] = coordinator
diag = await async_get_config_entry_diagnostics(hass, coordinator._entry)
assert len(diag['sub_units']) == 1
assert diag['sub_units'][0]['kind'] == 'prefixed'
# diagnostics reports the raw modelNum field verbatim (board revision/
# capability-bitmap suffix included), unlike device_info_for's model
# (split at '|') -- confirms it's still the wall unit's own identity,
# not the master's TP2X_FAC_BORA_21K.
assert diag['sub_units'][0]['model'].startswith('TP2X_FAC_BORA_RAC_21K')
assert diag['sub_units_skipped'] == []
+8
View File
@@ -17,6 +17,14 @@ class _FakeCoordinator:
def __init__(self, last_resources):
self.last_resources = last_resources
def canonical_resources(self, sub_unit):
# Every bound entity in this test file uses the default MAIN
# sub-unit (identity transform), so the canonical view is just the
# raw snapshot -- same shape as the real
# LocalThingsCoordinator.canonical_resources for a device with no
# sub-units (issue #177).
return self.last_resources
def _bound(desc, href):
capability = Capability(href=href, entities=(desc,))
+54
View File
@@ -859,6 +859,40 @@ def test_registry_reproduces_golden_state_keys_for_airconditioner_fac_bora():
)
def _new_sub_unit_aware_state_keys(name):
"""Like _new_state_keys, but runs the full sub-unit-aware pipeline
(enumerate_sub_units + discover_partitioned, issue #177) instead of a
single discover() call, so the golden for a composite-device fixture
captures every materialized unit's keys (unit1_-/sub_<uuid>_-prefixed),
not just the master's."""
from custom_components.localthings.registry.adapter import flatten
from tests.conftest import _discover_full, _load_device_full
resources, oic_res, seeds = _load_device_full(name)
bound, _materialized, _skipped, full_resources, _device_type_name = _discover_full(
resources, oic_res, seeds,
)
state = flatten(bound, full_resources)
return sorted(state.keys())
def test_registry_reproduces_golden_state_keys_for_airconditioner_artik051_dongle_fac_18k():
"""HJcom's ARTIK051_DONGLE_FAC_18K (issue #177, Pattern A -- indexed
siblings): a real v0.16.0 dump with a genuine second indoor unit at
`/device/1` (unit1_-prefixed keys below) and an unused SmartThings slot
at `/device/2` that answers its seed but never produces a materialized
unit (see DESIGN-177.md section 4 and test_subdevice_discovery.py's
explicit "/device/2 produces no entities" assertion) -- so this golden
has no `unit2_`-prefixed keys at all, which is the point."""
name = 'airconditioner_artik051_dongle_fac_18k'
golden = json.loads((GOLDEN / f'{name}.json').read_text())
state_keys = _new_sub_unit_aware_state_keys(name)
assert set(state_keys) == set(golden['state_keys']), (
f"state_keys mismatch:\n"
f" extra: {sorted(set(state_keys) - set(golden['state_keys']))}\n"
f" missing: {sorted(set(golden['state_keys']) - set(state_keys))}"
)
def test_registry_reproduces_golden_state_keys_for_airconditioner_cac():
"""TP1X_DA-AC-CAC-01001_0000 (issue #191) -- fell back to 'unknown' in
0.16.0 when oneUiVersion detection was dropped, since 'CAC' had never
@@ -879,6 +913,26 @@ def test_registry_reproduces_golden_state_keys_for_airconditioner_cac():
)
def test_registry_reproduces_golden_state_keys_for_airconditioner_fac_bora_2in1():
"""jhkwon19's TP2X_FAC_BORA_21K (issue #177, Pattern B -- UUID-prefixed
tree): device0/oic_res are real; the wall-mounted sub-unit's own
/information/vs/0 is real (confirmed live by the reporter), the rest of
its seed tree is constructed (see the fixture's own seeds_note) -- just
enough to bind a real climate card under the
`sub_6c2dff6dee5cdad16a5e000000000001_` prefix below. Distinct from
tests/fixtures/airconditioner_fac_bora_device.json, which is
deliberately left unchanged as the redacted-subdeviceIdList regression
case (zero sub-units)."""
name = 'airconditioner_fac_bora_2in1'
golden = json.loads((GOLDEN / f'{name}.json').read_text())
state_keys = _new_sub_unit_aware_state_keys(name)
assert set(state_keys) == set(golden['state_keys']), (
f"state_keys mismatch:\n"
f" extra: {sorted(set(state_keys) - set(golden['state_keys']))}\n"
f" missing: {sorted(set(golden['state_keys']) - set(state_keys))}"
)
def test_registry_reproduces_golden_state_keys_for_air_purifier_avt_ww():
"""AVT-WW-TP1-23-AXX500 (issue #190) -- next-gen BESPOKE Cube Air board;
reports device_type 'unknown' with empty oneUiVersion because 'VTWW' as a
+1 -33
View File
@@ -32,10 +32,7 @@ def test_read_identity_tolerates_missing_resources():
assert ident.model == ''
assert ident.serial is None
assert ident.device_types == ()
assert ident.raw == {
'/oic/p': {}, '/oic/d': {}, '/oic/res': [],
'/device/1': {}, '/device/2': {},
}
assert ident.raw == {'/oic/p': {}, '/oic/d': {}, '/oic/res': []}
def test_read_identity_captures_oic_d_device_types():
@@ -111,32 +108,3 @@ def test_read_identity_tolerates_malformed_oic_res():
FakeSession({('oic', 'res'): {'not': 'a list'}}), serial=None
)
assert ident.raw['/oic/res'] == []
def test_read_identity_probes_device_1_and_2():
"""/oic/res won't reveal a second logical Device's Collection on this
firmware family (see test_read_identity_captures_oic_res_links), so
/device/1 and /device/2 are probed directly -- same [devcol-rep,
{href, rep}, ...] batch shape /device/0 itself returns, parsed the same
way (parse_device0_batch)."""
sess = FakeSession({
('device', '1'): [
{'rt': ['x.com.samsung.devcol', 'oic.wk.col']},
{'href': '/power/vs/0', 'rep': {'x.com.samsung.da.power': 'On'}},
],
})
ident = read_identity(sess, serial=None)
assert ident.raw['/device/1'] == {
'/power/vs/0': {'x.com.samsung.da.power': 'On'},
}
# /device/2 was never in the table -> 4.04 -> tolerated absence.
assert ident.raw['/device/2'] == {}
def test_read_identity_tolerates_malformed_device_probe():
"""A bare Property map instead of a batch array (or anything else
non-list-shaped) must not explode."""
ident = read_identity(
FakeSession({('device', '1'): {'not': 'a batch'}}), serial=None
)
assert ident.raw['/device/1'] == {}
+6
View File
@@ -19,6 +19,12 @@ class _FakeCoordinator:
def resource(self, href):
return self.last_resources.get(href, {})
def canonical_resources(self, sub_unit):
# Every bound entity in this test uses the default MAIN
# sub-unit, so the canonical view is just the raw snapshot
# (issue #177 -- see LocalThingsEntity._resources).
return self.last_resources
async def async_send_command(self, bound, payload):
self.commands.append((bound, payload))
+6
View File
@@ -14,6 +14,12 @@ class _FakeCoordinator:
def __init__(self, last_resources):
self.last_resources = last_resources
def canonical_resources(self, sub_unit):
# Every entity built by _make_select uses the default MAIN
# sub-unit, so the canonical view is just the raw snapshot
# (issue #177 -- see LocalThingsEntity._resources).
return self.last_resources
def _make_select(desc, href, last_resources):
capability = Capability(href=href, entities=(desc,))
+220
View File
@@ -0,0 +1,220 @@
"""End-to-end discovery tests for issue #177's two composite-device
fixtures, against the real LocalThingsCoordinator (not the HA-free
registry-level helpers test_subunits.py/test_unique_ids.py use) -- this is
what actually exercises _enumerate_sub_units_blocking + _run_discovery
together, including device_info_for/via_device and the "no phantom
/device/2 entities" guarantee.
"""
from __future__ import annotations
import pytest
from homeassistant.core import HomeAssistant
from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.localthings.const import (
CONF_HOST, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, CONF_PORT, DOMAIN,
)
from custom_components.localthings.coordinator import LocalThingsCoordinator
from custom_components.localthings.registry.entities import ClimateDesc
from custom_components.localthings.registry.identity import DeviceIdentity
from tests.conftest import FakeCoapSession, _load_device_full
ENTRY_DATA = {
CONF_HOST: '10.0.0.177',
CONF_PORT: 49154,
CONF_LEAF_CERT_PEM: '-----BEGIN CERTIFICATE-----\nTEST-LEAF\n-----END CERTIFICATE-----',
CONF_LEAF_KEY_PEM: '-----BEGIN PRIVATE KEY-----\nTEST-LEAF-KEY\n-----END PRIVATE KEY-----',
}
def _coordinator(hass: HomeAssistant) -> LocalThingsCoordinator:
entry = MockConfigEntry(
domain=DOMAIN, data=ENTRY_DATA, unique_id='localthings_SUBDEVICE-TEST',
)
entry.add_to_hass(hass)
return LocalThingsCoordinator(hass, entry)
async def _discover(coordinator: LocalThingsCoordinator, name: str) -> None:
"""Run the same two-step sequence _async_update_data's first cycle does
(enumerate, then discover) against fixture data, without the polling/
reconnect machinery around it -- see coordinator.py's
_enumerate_sub_units_blocking/_run_discovery."""
resources, oic_res, seeds = _load_device_full(name)
coordinator._session = FakeCoapSession(seeds)
# _connect_session (skipped here -- the session is pre-set) is what
# normally populates _identity via read_identity; set it directly with
# the fixture's real /oic/res so enumeration sees the same links a live
# read_identity call would have captured.
coordinator._identity = DeviceIdentity(
manufacturer='Samsung Electronics', model='', name='', serial=None,
device_types=(), raw={'/oic/p': {}, '/oic/d': {}, '/oic/res': oic_res},
)
merged = await coordinator.hass.async_add_executor_job(
coordinator._enumerate_sub_units_blocking, resources,
)
# Mirror _async_update_data's first-cycle order exactly: discover, then
# drop the candidates the liveness gate rejected, then apply what's left
# to the observe/cache layer. The apply has to happen (canonical_resources
# -- device_info_for, is_legacy_board, ... -- reads the cache, not the
# dict passed to _run_discovery), but it has to happen *after* the gate,
# or a rejected slot's reps get frozen into the cache forever. Applying
# first here would leave this helper testing an ordering production no
# longer uses.
coordinator._run_discovery(merged)
for href, rep in coordinator._live_unit_resources(merged).items():
coordinator._observe.apply(href, rep, source='poll')
def _climate_bound(coordinator, sub_unit_key: str):
from custom_components.localthings.registry.subunits import MAIN
for b in coordinator.bound:
if isinstance(b.desc, ClimateDesc):
if sub_unit_key is None and b.sub_unit == MAIN:
return b
if sub_unit_key is not None and b.sub_unit.key == sub_unit_key:
return b
return None
# ---------------------------------------------------------------------------
# HJcom -- ARTIK051_DONGLE_FAC_18K, Pattern A (indexed siblings)
# ---------------------------------------------------------------------------
async def test_hjcom_materializes_master_and_bedroom_unit(hass: HomeAssistant):
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_artik051_dongle_fac_18k')
assert [su.key for su in coordinator.sub_units] == ['1']
main_climate = _climate_bound(coordinator, None)
unit1_climate = _climate_bound(coordinator, '1')
assert main_climate is not None
assert unit1_climate is not None
assert main_climate.href == '/mode/vs/0'
assert unit1_climate.href == '/mode/vs/1'
async def test_hjcom_device_2_produces_no_entities_at_all(hass: HomeAssistant):
"""HJcom's /device/2 is the unused SmartThings slot (DESIGN-177.md
section 4): it answers its seed with a full-shaped batch, but every
climate-state rep on it is empty. It must be recorded as skipped, not
materialized, and must contribute zero bound entities."""
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_artik051_dongle_fac_18k')
assert '2' not in [su.key for su in coordinator.sub_units]
assert any(
skip.sub_unit.kind == 'indexed' and skip.sub_unit.key == '2'
for skip in coordinator._skipped_sub_units
)
assert not any(b.sub_unit.key == '2' for b in coordinator.bound)
assert not any(href.endswith('/2') for href in coordinator._hot_hrefs)
assert not any(href.endswith('/2') for href in coordinator._warm_hrefs)
async def test_hjcom_unit1_device_info_links_via_device_to_master(hass: HomeAssistant):
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_artik051_dongle_fac_18k')
unit1 = next(su for su in coordinator.sub_units if su.key == '1')
info = coordinator.device_info_for(unit1)
master_serial = coordinator.device_serial
assert info['identifiers'] == {(DOMAIN, f'{master_serial}_1')}
assert info['via_device'] == (DOMAIN, master_serial)
# The sub-unit's own /information/vs/1 (real, ARTIK051_DONGLE_FAC_RAC_18K)
# is what names/models this device, not the master's.
assert info['model'] == 'ARTIK051_DONGLE_FAC_RAC_18K'
# ---------------------------------------------------------------------------
# jhkwon19 -- TP2X_FAC_BORA_21K, Pattern B (UUID-prefixed tree)
# ---------------------------------------------------------------------------
_SUB_UUID = '6c2dff6d-ee5c-dad1-6a5e-000000000001'
async def test_fac_bora_2in1_materializes_prefixed_wall_unit(hass: HomeAssistant):
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_fac_bora_2in1')
assert [su.key for su in coordinator.sub_units] == [_SUB_UUID]
assert coordinator.sub_units[0].kind == 'prefixed'
main_climate = _climate_bound(coordinator, None)
sub_climate = _climate_bound(coordinator, _SUB_UUID)
assert main_climate is not None
assert sub_climate is not None
assert main_climate.href == '/mode/vs/0'
assert sub_climate.href == f'/{_SUB_UUID}/mode/vs/0'
async def test_fac_bora_2in1_sub_unit_device_info(hass: HomeAssistant):
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_fac_bora_2in1')
unit = coordinator.sub_units[0]
info = coordinator.device_info_for(unit)
master_serial = coordinator.device_serial
assert info['identifiers'] == {(DOMAIN, f'{master_serial}_{_SUB_UUID}')}
assert info['via_device'] == (DOMAIN, master_serial)
# Confirmed live by the reporter (DESIGN-177.md section 1): the wall
# unit's own identity, distinct from the master's TP2X_FAC_BORA_21K.
assert info['model'] == 'TP2X_FAC_BORA_RAC_21K'
async def test_fac_bora_2in1_unique_ids_include_sub_prefix(hass: HomeAssistant):
"""The prefixed unit's unique_id carries the full subdevice UUID
(non-alphanumerics stripped), not a truncation or an ordinal -- see
SubUnit.key_prefix."""
coordinator = _coordinator(hass)
await _discover(coordinator, 'airconditioner_fac_bora_2in1')
from custom_components.localthings.entity import LocalThingsEntity
sub_climate = _climate_bound(coordinator, _SUB_UUID)
entity = LocalThingsEntity(coordinator, sub_climate)
expected_slug = _SUB_UUID.replace('-', '')
assert entity._attr_unique_id == (
f"{DOMAIN}_{coordinator.device_serial}_sub_{expected_slug}_climate"
)
async def test_multidevice_probe_never_reaches_discovery_or_the_cache(
hass: HomeAssistant,
):
"""/multidevice/vs/0 is probed on every device but is metadata, not state.
It has to stay out of the resources dict on both counts. Discovery would
otherwise report it as an unbound href on every family whose registry
doesn't ignore that path -- only the AC one does -- raising a spurious
"incomplete capability coverage" repair for, say, a washer whose
firmware happens to answer it. And nothing polls it after discovery, so
anything applied to the state cache would sit frozen there forever.
Driven with a washer fixture precisely because the AC registry's own
ignore entry would mask the coverage half of this on an AC.
"""
resources, _oic, _seeds = _load_device_full('washer_flexwash')
coordinator = _coordinator(hass)
coordinator._session = FakeCoapSession({
'/multidevice/vs/0': {'x.com.samsung.da.numofsubdevice': '2'},
})
coordinator._identity = DeviceIdentity(
manufacturer='Samsung Electronics', model='', name='', serial=None,
device_types=(), raw={'/oic/p': {}, '/oic/d': {}, '/oic/res': []},
)
merged = await hass.async_add_executor_job(
coordinator._enumerate_sub_units_blocking, resources,
)
coordinator._run_discovery(merged)
for href, rep in coordinator._live_unit_resources(merged).items():
coordinator._observe.apply(href, rep, source='poll')
assert '/multidevice/vs/0' not in merged
assert '/multidevice/vs/0' not in coordinator._unbound_hrefs
assert '/multidevice/vs/0' not in coordinator.last_resources
# Still captured, just not as device state.
assert coordinator._multidevice == {'x.com.samsung.da.numofsubdevice': '2'}
+581
View File
@@ -0,0 +1,581 @@
"""Tests for registry/subunits.py -- multi-indoor-unit ("composite device")
support, issue #177. See DESIGN-177.md for the two board patterns
(`ARTIK051_DONGLE_FAC_18K`'s indexed siblings, `TP2X_FAC_BORA_21K`'s
UUID-prefixed tree) this module unifies.
"""
from __future__ import annotations
import cbor2
from custom_components.localthings.registry.capability import Capability
from custom_components.localthings.registry.entities import BinarySensorDesc, SensorDesc
from custom_components.localthings.registry.subunits import (
MAIN, SubUnit, canonical_view, discover_partitioned, enumerate_sub_units,
normalize_seed_batch,
)
_UUID = '6c2dff6d-ee5c-dad1-6a5e-000000000001'
def _indexed(n: str) -> SubUnit:
return SubUnit(kind='indexed', key=n, seed_path=('device', n))
def _prefixed(sub_id: str) -> SubUnit:
return SubUnit(kind='prefixed', key=sub_id, seed_path=(sub_id, 'device', '0'))
# ---------------------------------------------------------------------------
# SubUnit.to_actual / to_canonical / owns / key_prefix
# ---------------------------------------------------------------------------
class TestMainIsIdentity:
def test_to_actual_unchanged(self):
assert MAIN.to_actual('/mode/vs/0') == '/mode/vs/0'
def test_to_canonical_unchanged(self):
assert MAIN.to_canonical('/mode/vs/0') == '/mode/vs/0'
def test_owns_is_always_false(self):
"""MAIN never "owns" a href by this definition -- it gets whatever's
left after every other sub-unit's hrefs are excluded (canonical_view)."""
assert MAIN.owns('/mode/vs/0') is False
assert MAIN.owns('/mode/vs/1') is False
def test_key_prefix_is_empty(self):
"""The master unit's flattened state keys must stay byte-identical
to every device shipped before issue #177 -- see adapter._key."""
assert MAIN.key_prefix == ''
class TestIndexedTransform:
def test_to_actual_rewrites_trailing_zero(self):
unit = _indexed('1')
assert unit.to_actual('/mode/vs/0') == '/mode/vs/1'
assert unit.to_actual('/device/0') == '/device/1'
def test_to_actual_leaves_non_zero_trailing_segment_alone(self):
"""Deliberately not a "replace any trailing digit" rule -- a genuine
multi-instance resource (the fridge's pattern-cap hrefs, e.g.
'/door/vs/1') must not be misread as a sub-unit's own href."""
unit = _indexed('1')
assert unit.to_actual('/door/vs/1') == '/door/vs/1'
def test_to_canonical_round_trips(self):
unit = _indexed('1')
assert unit.to_canonical(unit.to_actual('/mode/vs/0')) == '/mode/vs/0'
def test_to_canonical_rejects_wrong_index(self):
unit = _indexed('1')
assert unit.to_canonical('/mode/vs/2') is None
def test_to_canonical_rejects_index_zero(self):
"""Index 0 belongs to MAIN, never to an indexed sub-unit."""
unit = _indexed('1')
assert unit.to_canonical('/mode/vs/0') is None
def test_owns(self):
unit = _indexed('1')
assert unit.owns('/mode/vs/1') is True
assert unit.owns('/mode/vs/0') is False
assert unit.owns('/mode/vs/2') is False
def test_key_prefix(self):
assert _indexed('1').key_prefix == 'unit1_'
assert _indexed('2').key_prefix == 'unit2_'
class TestPrefixedTransform:
def test_to_actual_prepends_id(self):
unit = _prefixed(_UUID)
assert unit.to_actual('/mode/vs/0') == f'/{_UUID}/mode/vs/0'
def test_to_canonical_round_trips(self):
unit = _prefixed(_UUID)
assert unit.to_canonical(unit.to_actual('/mode/vs/0')) == '/mode/vs/0'
def test_to_canonical_rejects_missing_prefix(self):
unit = _prefixed(_UUID)
assert unit.to_canonical('/mode/vs/0') is None
def test_to_canonical_requires_path_boundary(self):
"""A different, longer id that merely starts with the same
characters must not be mistaken for this one's own href."""
unit = _prefixed(_UUID)
assert unit.to_canonical(f'/{_UUID}extra/mode/vs/0') is None
def test_owns(self):
unit = _prefixed(_UUID)
assert unit.owns(f'/{_UUID}/mode/vs/0') is True
assert unit.owns('/mode/vs/0') is False
def test_key_prefix_strips_non_alphanumerics(self):
assert _prefixed(_UUID).key_prefix == 'sub_6c2dff6dee5cdad16a5e000000000001_'
# ---------------------------------------------------------------------------
# canonical_view
# ---------------------------------------------------------------------------
class TestCanonicalView:
def test_main_with_no_sub_units_is_unchanged(self):
resources = {'/mode/vs/0': {'a': 1}, '/power/vs/0': {'b': 2}}
assert canonical_view(MAIN, resources, []) == resources
def test_main_excludes_sub_unit_owned_hrefs(self):
"""A sibling's own /mode/vs/1 must not leak into the master's
canonical /mode/vs/0 view -- otherwise exists_fn/is_legacy_board
checks that scan the whole dict would see two units' state mixed
together under one key."""
resources = {
'/mode/vs/0': {'unit': 'main'},
'/mode/vs/1': {'unit': 'sibling'},
}
unit1 = _indexed('1')
view = canonical_view(MAIN, resources, [unit1])
assert view == {'/mode/vs/0': {'unit': 'main'}}
def test_indexed_view_is_rewritten_to_canonical_hrefs(self):
resources = {
'/mode/vs/0': {'unit': 'main'},
'/mode/vs/1': {'unit': 'sibling'},
'/power/vs/1': {'p': 'On'},
}
unit1 = _indexed('1')
view = canonical_view(unit1, resources, [unit1])
assert view == {
'/mode/vs/0': {'unit': 'sibling'},
'/power/vs/0': {'p': 'On'},
}
def test_prefixed_view_is_rewritten_to_canonical_hrefs(self):
resources = {
f'/{_UUID}/mode/vs/0': {'unit': 'sub'},
'/mode/vs/0': {'unit': 'main'},
}
unit = _prefixed(_UUID)
view = canonical_view(unit, resources, [unit])
assert view == {'/mode/vs/0': {'unit': 'sub'}}
def test_including_main_in_sub_units_is_harmless(self):
"""MAIN.owns() is always False, so passing the full roster
(including MAIN itself) to canonical_view must not change anything."""
resources = {'/mode/vs/0': {'unit': 'main'}, '/mode/vs/1': {'unit': 'sib'}}
unit1 = _indexed('1')
assert (canonical_view(MAIN, resources, [MAIN, unit1])
== canonical_view(MAIN, resources, [unit1]))
# ---------------------------------------------------------------------------
# normalize_seed_batch
# ---------------------------------------------------------------------------
def test_normalize_seed_batch_indexed_is_a_no_op():
unit = _indexed('1')
batch = {'/mode/vs/1': {'a': 1}}
assert normalize_seed_batch(unit, batch) == batch
def test_normalize_seed_batch_prefixed_adds_missing_prefix():
unit = _prefixed(_UUID)
batch = {'/mode/vs/0': {'a': 1}}
assert normalize_seed_batch(unit, batch) == {f'/{_UUID}/mode/vs/0': {'a': 1}}
def test_normalize_seed_batch_prefixed_leaves_already_prefixed_alone():
unit = _prefixed(_UUID)
batch = {f'/{_UUID}/mode/vs/0': {'a': 1}}
assert normalize_seed_batch(unit, batch) == batch
# ---------------------------------------------------------------------------
# enumerate_sub_units
# ---------------------------------------------------------------------------
class _FakeSession:
"""path-tuple -> raw batch list, cbor-encoded on GET -- same shape as
test_identity.py's FakeSession, extended to serve Collection batches
(a CBOR list), not just a bare Property map."""
def __init__(self, table):
self.table = table
def get(self, path, timeout=10.0):
body = self.table.get(tuple(path))
if body is None:
return 0x84, b''
return 0x45, cbor2.dumps(body)
_DEVCOL_REP = {'rt': ['x.com.samsung.devcol', 'oic.wk.col']}
def test_enumerate_indexed_from_oic_res_links():
oic_res = [{'di': 'aaaa', 'links': [
{'href': '/device/0'}, {'href': '/device/1'}, {'href': '/device/2'},
]}]
sess = _FakeSession({
('device', '1'): [_DEVCOL_REP, {'href': '/mode/vs/1', 'rep': {'m': 1}}],
('device', '2'): [_DEVCOL_REP, {'href': '/mode/vs/2', 'rep': {'m': 2}}],
})
units, extra = enumerate_sub_units(sess, {}, oic_res)
assert sorted((u.kind, u.key) for u in units) == [
('indexed', '1'), ('indexed', '2'),
]
assert extra == {'/mode/vs/1': {'m': 1}, '/mode/vs/2': {'m': 2}}
def test_enumerate_indexed_falls_back_to_speculative_probe_when_oic_res_hides_the_tree():
"""A board whose /oic/res doesn't reveal a second logical Device (e.g.
it hides the whole tree, TP2X_FAC_BORA_21K-style) falls back to probing
/device/1 and /device/2 directly -- the same bound the old
identity.py-level probe used."""
sess = _FakeSession({
('device', '1'): [_DEVCOL_REP, {'href': '/mode/vs/1', 'rep': {'m': 1}}],
# /device/2 not in the table -> 4.04 -> not materialized.
})
units, extra = enumerate_sub_units(sess, {}, oic_res_links=[])
assert [(u.kind, u.key) for u in units] == [('indexed', '1')]
assert extra == {'/mode/vs/1': {'m': 1}}
def test_enumerate_indexed_only_materializes_units_with_a_non_empty_batch():
"""Per issue #177's design call: any seed that answers with a non-empty
batch is materialized, but an empty ({}) answer is not -- no separate
"is this unit real" gate."""
sess = _FakeSession({}) # neither /device/1 nor /device/2 answers
units, extra = enumerate_sub_units(sess, {}, oic_res_links=[])
assert units == []
assert extra == {}
def test_enumerate_prefixed_from_subdevice_id_list():
resources = {
'/subdevices/vs/0': {'x.com.samsung.da.subdeviceIdList': [_UUID]},
}
sess = _FakeSession({
(_UUID, 'device', '0'): [
_DEVCOL_REP, {'href': '/mode/vs/0', 'rep': {'m': 'cool'}},
],
})
units, extra = enumerate_sub_units(sess, resources, oic_res_links=[])
assert [(u.kind, u.key) for u in units] == [('prefixed', _UUID)]
# Batch echoed the bare (unprefixed) href -- normalized to carry the id.
assert extra == {f'/{_UUID}/mode/vs/0': {'m': 'cool'}}
def test_enumerate_prefixed_tolerates_redacted_string_id_list():
"""subdeviceIdList matches redact.py's 'deviceid' substring rule, and the
real airconditioner_fac_bora_device.json fixture carries the literal
redacted string there -- this must yield zero sub-units, not crash."""
resources = {
'/subdevices/vs/0': {'x.com.samsung.da.subdeviceIdList': 'REDACTED'},
}
units, extra = enumerate_sub_units(_FakeSession({}), resources, oic_res_links=[])
assert units == []
assert extra == {}
def test_enumerate_no_subdevices_resource_at_all():
units, extra = enumerate_sub_units(_FakeSession({}), {}, oic_res_links=[])
assert units == []
assert extra == {}
def test_enumerate_indexed_materializes_candidate_regardless_of_content():
"""enumerate_sub_units itself has no way to tell a real sibling from an
unused slot that merely answers the same shape (HJcom's /device/2) --
that's discover_partitioned's job (see its own tests below). A
same-shaped batch of otherwise-empty reps is still returned as a
candidate here."""
oic_res = [{'di': 'a', 'links': [{'href': '/device/2'}]}]
sess = _FakeSession({
('device', '2'): [_DEVCOL_REP, {'href': '/power/vs/2', 'rep': {}},
{'href': '/configuration/vs/2', 'rep': {'region': '123'}}],
})
units, extra = enumerate_sub_units(sess, {}, oic_res)
assert [(u.kind, u.key) for u in units] == [('indexed', '2')]
assert extra == {'/power/vs/2': {}, '/configuration/vs/2': {'region': '123'}}
def test_enumerate_probe_log_reports_every_attempt():
oic_res = [{'di': 'aaaa', 'links': [{'href': '/device/1'}]}]
sess = _FakeSession({
('device', '1'): [_DEVCOL_REP, {'href': '/mode/vs/1', 'rep': {'m': 1}}],
})
probes: dict[str, bool] = {}
enumerate_sub_units(sess, {}, oic_res, probe_log=probes.__setitem__)
# /multidevice/vs/0 is always probed too (issue #177 follow-up) -- not
# in this session's table, so it reports "checked, nothing there".
assert probes == {'/device/1': True, '/multidevice/vs/0': False}
def test_enumerate_probe_log_reports_empty_answers_too():
"""Distinguishes "checked, nothing there" from "never checked" (the
posture the speculative-probe code this replaced used to document
directly in identity.py)."""
probes: dict[str, bool] = {}
enumerate_sub_units(_FakeSession({}), {}, oic_res_links=[], probe_log=probes.__setitem__)
assert probes == {
'/device/1': False, '/device/2': False, '/multidevice/vs/0': False,
}
def test_enumerate_multidevice_vs_0_is_captured_for_diagnostics_not_a_gate():
"""/multidevice/vs/0 (issue #177 follow-up) is corroborating evidence
only -- captured into the merged resources so diagnostics can report
it, never consulted by the liveness gate itself."""
sess = _FakeSession({
('multidevice', 'vs', '0'): {'x.com.samsung.da.numofsubdevice': '2'},
})
units, extra = enumerate_sub_units(sess, {}, oic_res_links=[])
assert units == [] # no /device/<n> or subdevice id answered
assert extra == {
'/multidevice/vs/0': {'x.com.samsung.da.numofsubdevice': '2'},
}
def test_enumerate_both_patterns_checked_independently():
"""No board needs disambiguation logic -- enumeration just checks both
signals and takes whatever answers."""
resources = {
'/subdevices/vs/0': {'x.com.samsung.da.subdeviceIdList': [_UUID]},
}
oic_res = [{'di': 'a', 'links': [{'href': '/device/1'}]}]
sess = _FakeSession({
(_UUID, 'device', '0'): [_DEVCOL_REP, {'href': '/mode/vs/0', 'rep': {'m': 1}}],
('device', '1'): [_DEVCOL_REP, {'href': '/mode/vs/1', 'rep': {'m': 2}}],
})
units, extra = enumerate_sub_units(sess, resources, oic_res)
assert sorted((u.kind, u.key) for u in units) == [
('indexed', '1'), ('prefixed', _UUID),
]
# ---------------------------------------------------------------------------
# discover_partitioned
# ---------------------------------------------------------------------------
class _FakeRegistry:
def __init__(self, name, capabilities, pattern_capabilities=()):
self.name = name
self.capabilities = capabilities
self.pattern_capabilities = list(pattern_capabilities)
def test_discover_partitioned_binds_main_and_sub_unit_separately():
mode_cap = Capability(
href='/mode/vs/0',
entities=(BinarySensorDesc(key='mode', field='m'),),
)
registry = {'/mode/vs/0': [mode_cap]}
reg = _FakeRegistry('airconditioner', registry)
unit1 = _indexed('1')
resources = {
'/mode/vs/0': {'m': 'main'},
'/mode/vs/1': {'m': 'sibling'},
}
def resolve(_resources):
return reg
bound, device_type_name, materialized, skipped = discover_partitioned(
resources, [unit1], resolve, fallback_capabilities={},
)
assert device_type_name == 'airconditioner'
assert materialized == [unit1]
assert skipped == []
by_href = {b.href: b for b in bound}
assert set(by_href) == {'/mode/vs/0', '/mode/vs/1'}
assert by_href['/mode/vs/0'].sub_unit == MAIN
assert by_href['/mode/vs/1'].sub_unit == unit1
def test_discover_partitioned_main_pass_excludes_sub_unit_hrefs_from_unbound():
"""A sub-unit's own /mode/vs/1 must not land in the main pass's unbound
list -- otherwise it raises a spurious coverage-gap repair for a href
nothing in the main registry claims literally, even though the *same*
canonical /mode/vs/0 href is properly claimed for both units.
A registry that binds /mode/vs/0 (so both the main pass and the
sub-unit pass over their own canonical views have something to claim
it) makes this unambiguous: if partitioning were broken -- e.g. the
main pass iterating unfiltered `resources` instead of `main_view` --
/mode/vs/1 would show up in `unbound` because nothing in the registry
is keyed on the literal string '/mode/vs/1'. With correct partitioning
the main pass never sees that href at all (canonical_view excludes it),
and the sub-unit pass resolves it via its own canonical '/mode/vs/0'."""
cap = Capability(href='/mode/vs/0', entities=(BinarySensorDesc(key='mode', field='m'),))
reg = _FakeRegistry('airconditioner', {'/mode/vs/0': [cap]})
unit1 = _indexed('1')
resources = {
'/mode/vs/0': {'m': 'main'},
'/mode/vs/1': {'m': 'sibling'},
}
unbound = []
discover_partitioned(
resources, [unit1], lambda r: reg, fallback_capabilities={},
log=unbound.append,
)
assert unbound == []
def test_discover_partitioned_sub_unit_resolves_its_own_registry():
"""A sub-unit reporting its own /information/vs/0 resolves its own
device type (jhkwon19's wall unit: TP2X_FAC_BORA_RAC_21K -> RAC ->
airconditioner) independent of the master's."""
main_cap = Capability(href='/mode/vs/0', entities=(BinarySensorDesc(key='m', field='x'),))
sub_cap = Capability(href='/mode/vs/0', entities=(BinarySensorDesc(key='m2', field='x'),))
main_reg = _FakeRegistry('main_type', {'/mode/vs/0': [main_cap]})
sub_reg = _FakeRegistry('sub_type', {'/mode/vs/0': [sub_cap]})
unit1 = _indexed('1')
resources = {'/mode/vs/0': {'x': 1}, '/mode/vs/1': {'x': 2}}
def resolve(view):
# The sub-unit's canonical view is exactly {'/mode/vs/0': {'x': 2}}.
return sub_reg if view.get('/mode/vs/0', {}).get('x') == 2 else main_reg
bound, device_type_name, materialized, skipped = discover_partitioned(
resources, [unit1], resolve, fallback_capabilities={},
)
assert device_type_name == 'main_type'
assert materialized == [unit1]
assert skipped == []
keys = {(b.href, b.desc.key) for b in bound}
assert ('/mode/vs/0', 'm') in keys
assert ('/mode/vs/1', 'm2') in keys
def test_discover_partitioned_sub_unit_falls_back_to_master_registry():
"""A sub-unit that reports no identity of its own resolves through the
master's registry instead of the global fallback -- the master itself
must actually resolve here (it reports /information/vs/0), or there is
no master registry for the fallback to reach for."""
cap = Capability(href='/mode/vs/0', entities=(BinarySensorDesc(key='m', field='x'),))
main_reg = _FakeRegistry('airconditioner', {'/mode/vs/0': [cap]})
unit1 = _indexed('1')
resources = {
'/information/vs/0': {'model': 'main'},
'/mode/vs/0': {'x': 1},
'/mode/vs/1': {'x': 2},
}
def resolve(view):
return main_reg if view.get('/information/vs/0') else None
bound, _, materialized, skipped = discover_partitioned(
resources, [unit1], resolve, fallback_capabilities={},
)
assert materialized == [unit1]
assert skipped == []
hrefs = {b.href for b in bound}
assert '/mode/vs/1' in hrefs
def test_discover_partitioned_no_sub_units_matches_plain_discover():
"""For a device with no sub-units this must be exactly the single
discover() call it replaces -- the hard regression guard the whole
design exists to protect."""
from custom_components.localthings.registry.discovery import discover
cap = Capability(href='/mode/vs/0', entities=(BinarySensorDesc(key='m', field='x'),))
reg = _FakeRegistry('airconditioner', {'/mode/vs/0': [cap]})
resources = {'/mode/vs/0': {'x': 1}}
bound_via_helper, _, materialized, skipped = discover_partitioned(
resources, [], lambda r: reg, fallback_capabilities={},
)
bound_direct = discover(resources, reg.capabilities, reg.pattern_capabilities)
assert bound_via_helper == bound_direct
assert materialized == []
assert skipped == []
# ---------------------------------------------------------------------------
# discover_partitioned's materialization gate: a candidate is only kept if
# it produced at least one *primary* (no entity_category) bound entity whose
# flattened value isn't None. This is what tells HJcom's real /device/1
# sibling apart from the unused /device/2 slot that answers the same shape.
# ---------------------------------------------------------------------------
def test_discover_partitioned_skips_candidate_with_no_live_primary_entity():
"""A candidate whose only populated entity is diagnostic-category
doesn't count -- exactly HJcom's /device/2 shape (an alarm_code
sensor reading something even though the unit itself is empty)."""
diag_cap = Capability(
href='/alarms/vs/0',
entities=(SensorDesc(key='alarm_code', field='code', entity_category='diagnostic'),),
)
reg = _FakeRegistry('airconditioner', {'/alarms/vs/0': [diag_cap]})
unit2 = _indexed('2')
resources = {
'/alarms/vs/0': {'code': 'main-alarm'},
'/alarms/vs/2': {'code': 'ErrorCode_OFF'}, # populated, but diagnostic-only
}
bound, _, materialized, skipped = discover_partitioned(
resources, [unit2], lambda r: reg, fallback_capabilities={},
)
assert materialized == []
assert len(skipped) == 1
assert skipped[0].sub_unit == unit2
assert skipped[0].hrefs == ('/alarms/vs/2',)
# The candidate contributes nothing at all -- not even its diagnostic
# entity -- once skipped.
assert all(b.sub_unit != unit2 for b in bound)
def test_discover_partitioned_materializes_candidate_with_live_primary_entity():
"""A candidate with a populated *primary* (no entity_category) entity
is materialized, even alongside an all-empty diagnostic sibling href --
exactly HJcom's /device/1 shape."""
climate_cap = Capability(
href='/mode/vs/0',
entities=(BinarySensorDesc(key='mode', field='m'),), # no entity_category -> primary
)
diag_cap = Capability(
href='/alarms/vs/0',
entities=(SensorDesc(key='alarm_code', field='code', entity_category='diagnostic'),),
)
reg = _FakeRegistry(
'airconditioner', {'/mode/vs/0': [climate_cap], '/alarms/vs/0': [diag_cap]},
)
unit1 = _indexed('1')
resources = {
'/mode/vs/0': {'m': 'main'},
'/alarms/vs/0': {'code': 'main-alarm'},
'/mode/vs/1': {'m': 'Cool'},
'/alarms/vs/1': {}, # empty -- not what makes this unit live
}
bound, _, materialized, skipped = discover_partitioned(
resources, [unit1], lambda r: reg, fallback_capabilities={},
)
assert materialized == [unit1]
assert skipped == []
hrefs = {b.href for b in bound if b.sub_unit == unit1}
assert hrefs == {'/mode/vs/1', '/alarms/vs/1'}
def test_discover_partitioned_skipped_candidate_contributes_no_hot_warm_hrefs():
"""A skipped candidate's hrefs must not appear via tier_log either --
'no entities, no hot/warm hrefs' (the skip is total, not just
entity-level)."""
cap = Capability(href='/mode/vs/0', poll_tier='warm',
entities=(SensorDesc(key='alarm_code', field='code',
entity_category='diagnostic'),))
reg = _FakeRegistry('airconditioner', {'/mode/vs/0': [cap]})
unit2 = _indexed('2')
resources = {'/mode/vs/0': {'code': 'x'}, '/mode/vs/2': {'code': 'y'}}
tiers = []
discover_partitioned(
resources, [unit2], lambda r: reg, fallback_capabilities={},
tier_log=lambda href, tier: tiers.append(href),
)
assert '/mode/vs/2' not in tiers
+82
View File
@@ -0,0 +1,82 @@
"""Corpus-wide invariant: adapter._key must be unique across every bound
entity that would actually be registered as an HA entity, for every fixture
in the corpus -- issue #177's whole SubUnit/key_prefix design exists to
protect this (see DESIGN-177.md section 3/6). Run over the entire fixture
set, not just the two new sub-unit fixtures, so a future dump -- sub-unit-
capable or not -- exercises it automatically.
"""
from collections import Counter
import pytest
from custom_components.localthings.entity import _is_included
from custom_components.localthings.registry.adapter import _key
from custom_components.localthings.registry.entities import PLATFORM_OF
from tests.conftest import FIXTURES, _discover_full, _load_device_full
_FIXTURE_NAMES = sorted(
p.name[:-len('_device.json')] for p in FIXTURES.glob('*_device.json')
)
class _FakeCoordinator:
"""Just enough of LocalThingsCoordinator's surface for entity.py's
_is_included -- the same one-time entity-creation gate every platform's
async_setup_entry runs (see entity.py's own module docstring)."""
def __init__(self, resources: dict[str, dict], sub_units):
self.last_resources = resources
self._sub_units = list(sub_units)
def canonical_resources(self, sub_unit):
from custom_components.localthings.registry.subunits import canonical_view
return canonical_view(sub_unit, self.last_resources, self._sub_units)
@pytest.mark.parametrize('name', _FIXTURE_NAMES)
def test_key_is_unique_across_all_bound_entities(name):
"""`_key(b)` only has to be unique among entities `_is_included` would
actually register -- discover() alone can (deliberately) produce two
BoundEntity rows sharing a key on the same href when they're gated by
mutually-exclusive `exists_fn`s (see fridge.py's
REFRIGERATION_FALLBACK.defrost_active, which only exists when
DEFROST_BLOCK_STATUS's own `/defrost/block/vs/0` href is absent) --
only one of the two is ever actually included for a real device, so
checking the raw `bound` list would flag devices that have always
worked correctly. It also only has to be unique *within one HA
platform* -- unique_id collisions are scoped by (integration, platform)
in HA's entity registry (see entity.py's `_attr_unique_id`, which
doesn't itself encode the platform), and several devices in this corpus
deliberately bind the same key to the same href on two different
platforms discriminated the same way (e.g. a SwitchDesc and a
BinarySensorDesc both named 'power_switch' on /power/0).
"""
resources, oic_res, seeds = _load_device_full(name)
bound, materialized, skipped, full_resources, device_type_name = _discover_full(
resources, oic_res, seeds,
)
coordinator = _FakeCoordinator(full_resources, materialized)
included = [b for b in bound if _is_included(b, coordinator)]
keys = [(PLATFORM_OF[type(b.desc)], _key(b)) for b in included]
dupes = {k: n for k, n in Counter(keys).items() if n > 1}
assert not dupes, (
f"{name}: duplicate (platform, _key) values across {len(included)} "
f"included entities (materialized sub-units: "
f"{[su.key for su in materialized]}): {dupes}"
)
def test_sub_unit_capable_fixtures_actually_exercise_a_sub_unit():
"""A meta-check on the test above: if both sub-unit fixtures somehow
stopped materializing any sub-unit (a regression in enumeration or the
materialization gate), the corpus-wide uniqueness test above would keep
passing vacuously -- it never gets to check a single collision. Assert
the two fixtures this design added actually produce a materialized
sub-unit, so that silent-vacuous-pass failure mode is caught here
instead."""
for name in ('airconditioner_artik051_dongle_fac_18k',
'airconditioner_fac_bora_2in1'):
resources, oic_res, seeds = _load_device_full(name)
_, materialized, _, _, _ = _discover_full(resources, oic_res, seeds)
assert materialized, f"{name}: expected at least one materialized sub-unit"