feat(air_purifier): expose the AI Purify sensing engine on /airlevelcheck/vs/0

/airlevelcheck/vs/0 has been covered as "periodic air-quality sensing
scheduler plumbing" since the registry gained a coverage stub for it. Two
AVT-WW-TP1-23-AXX500 dumps (issues #84 and #190) show it is not plumbing: it
drives the feature the SmartThings app calls AI Purify, where the unit wakes
on a timer, samples the air, and optionally acts on the result. Every field
is named, none are opaque, and two of them are already user-set on the
reported units.

The select's three on-states are the app's own options rather than an
invented grouping -- it offers exactly "Sensing only" (sample, take no
action), "Auto clean" (purify while the air reads bad, stop once it
improves) and "Get notified" (raise a SmartThings notification). Labels were
transcribed from the Korean app and rendered in English; the auto-stop half
of "Auto clean" is the app's own description and is not otherwise visible in
the dump, which reports only the selected autoExeState. The remaining entity
names follow their raw fields rather than inventing a concept -- the skip
window is "sensing skip", after periodicSensingSkipStatus/Time.

Three of this registry's four board families report the resource with the
same field names -- TP1X_DA-AC-AIR (#130), A-VTWW-TP2 (#151) and AVT-WW-TP1
(#84, #190). Only ARTIK051_TVTL (#56) has no such href, and its golden is
unchanged. Bound unconditionally rather than behind a match_fn; the one field
that genuinely varies (periodicSensingInterval, absent on the #130 board) is
gated per-entity, so that board gets eight entities instead of nine rather
than a broken one.

range_hood.AIR_LEVEL_CHECK already models this same href, and its read-only
keys are reused verbatim here so both families share one catalog entry. It is
deliberately not imported: the hood exposes periodic_air_sensing as a
read-only BinarySensorDesc and this board needs a writable SwitchDesc on that
key, so reusing the hood's capability would migrate every hood user's entity
to a different platform.

Every write was exercised on AVT-WW-TP1-23-AXX500 hardware. This board hands
out 2.04 for writes it silently discards (see HEPA_FILTER's filter-reset
note), so an echo proves nothing -- each was judged by whether the value
survived a reconnect, which forces a new DTLS session, fresh discovery and a
fresh observe of the href, leaving no cached state to read back:

  * sensing_mode's combined two-field PUT lands both fields, both ways:
    sensing_only -> auto_purify raises autoExeState with activation still On,
    and back again lowers it.
  * The sensing-skip switch holds Off -> On and back.
  * The half-preserving time writes hold: from 13:00-23:00, writing
    start=07:30 then end=22:00 left the device on '07302200' -- each write
    kept the half it wasn't given.
  * periodic_air_sensing and sensing_interval: writing 60 s drove an observed
    ~60 s sensing cycle.
  * The read side of the skip window is separately cross-confirmed on two
    units: #84's sits at the inert '00000000', #190's carries a real
    '03002300' (03:00-23:00), which is what pins the HHMMHHMM split.
  * The other two families get the writes on field-shape grounds -- the same
    basis on which they already share MODE, HEPA_FILTER and the air-quality
    sensors.

range_hood._timestamp moves to common.epoch_to_utc so both callers share it,
matching how filter_usage_percent was shared. No behaviour change.

Every existing entity is untouched: the three golden updates are purely
additive, no renames, no unit or device_class changes.
This commit is contained in:
kkqq9320
2026-08-04 12:57:03 +09:00
parent 0c2e219464
commit 412fff9b99
13 changed files with 635 additions and 13 deletions
@@ -22,6 +22,12 @@ match_fn discriminators that keep them from colliding):
resource surface as A-VTWW-TP2-21-COMMON above; no new capabilities
needed.
AIR_LEVEL_CHECK ("AI Purify" -- the periodic air-quality sensing engine on
/airlevelcheck/vs/0) is shared by the last three of those: their dumps all
carry the resource with the same field names, and only the TVTL family has no
such href. It was covered as opaque plumbing until two AVT-WW-TP1 dumps
(issues #84 and #190) showed it drives a real user-facing feature.
Reuses dishwasher.DIAGNOSIS for /diagnosis/vs/0 (identical field/write
contract).
"""
@@ -38,6 +44,7 @@ REGISTRY = DeviceRegistry(
*common.POWER,
dishwasher.DIAGNOSIS,
air_purifier.AIR_QUALITY,
air_purifier.AIR_LEVEL_CHECK,
air_purifier.FILTER,
air_purifier.DEVICE_ACTIVE,
air_purifier.AIRFLOW_GENERIC,
@@ -58,6 +58,8 @@ one unit, stuck at 0 throughout on the other), so AIRFLOW_VS_FALLBACK below
stays a plain read-only diagnostic even after this change.
"""
import datetime
from ..capability import Capability
from ..entities import (
BinarySensorDesc,
@@ -66,8 +68,9 @@ from ..entities import (
SelectDesc,
SensorDesc,
SwitchDesc,
TimeDesc,
)
from .common import filter_usage_percent, int_or_none, sensor_item_value
from .common import epoch_to_utc, filter_usage_percent, int_or_none, sensor_item_value
from .laundry import bool_option_exists, bool_option_value, option_value, option_write
# Newer TP1X_DA-AC-AIR-class boards (e.g. TP1X_DA-AC-AIR-01031_0000, issue
@@ -507,6 +510,246 @@ SOUND_VOLUME = Capability(
),
)
# ---------------------------------------------------------------------------
# AI Purify -- /airlevelcheck/vs/0 (issues #84 and #190)
#
# Covered as "periodic air-quality sensing scheduler plumbing" until two dumps
# of the AVT-WW-TP1-23 board showed it is not plumbing: it drives the feature
# the SmartThings app calls AI Purify, where the unit wakes on a timer, samples
# the air, and optionally acts on the result. Every field is named, none are
# opaque, and two of them are already user-set on the reported units.
#
# Three of this registry's four board families report the resource with the
# same field names -- TP1X_DA-AC-AIR (#130), A-VTWW-TP2 (#151) and AVT-WW-TP1
# (#84, #190); only ARTIK051_TVTL (#56) has no such href. Bound unconditionally
# rather than behind a match_fn so any board reporting it is covered; the one
# field that genuinely varies is gated per-entity below.
#
# range_hood.AIR_LEVEL_CHECK already models this same href, and its read-only
# keys (air_sensing_state / last_air_sensing_time / last_air_sensing_level) are
# reused verbatim so both families share one catalog entry. It is deliberately
# NOT imported: the hood exposes periodic_air_sensing as a read-only
# BinarySensorDesc and this board needs a writable SwitchDesc on that same key,
# so reusing the hood's capability would migrate every hood user's entity to a
# different platform.
#
# Verification: every write below was exercised on AVT-WW-TP1-23-AXX500
# hardware. This board returns 2.04 for writes it silently discards (see
# HEPA_FILTER's filter-reset note), so an echo proves nothing -- each was
# judged by the value surviving a reconnect, which forces a new DTLS session,
# fresh discovery and a fresh observe of this href, leaving no cached state to
# read back. The other two families get the writes on field-shape grounds, the
# same basis on which they already share MODE, HEPA_FILTER and the air-quality
# sensors.
# ---------------------------------------------------------------------------
def _interval_minutes(seconds):
"""Device stores the interval in seconds; the entity is in minutes."""
secs = int_or_none(seconds)
return round(secs / 60) if secs else None
def _interval_write(payload, rep, href=None):
# Minutes in the UI -> seconds on the wire (scalar string). Arbitrary
# values are honoured: confirmed on hardware, where writing 60 s drove an
# observed sensing cycle every ~60 s. The 1-60 min range on the entity is a
# UI guard, not a device-advertised limit -- the resource carries no
# supported-range field.
return ["airlevelcheck", "vs", "0"], {
"x.com.samsung.da.periodicSensingInterval": str(round(float(payload) * 60))
}
def _sensing_mode(rep):
"""Fold the periodic-sensing toggle and the auto-action into one status.
The three on-states are the SmartThings app's own options for this
feature, not an invented grouping -- the app offers exactly:
sensing_only "Sensing only" -- sample the air, take no action
auto_purify "Auto clean" -- run purification while the air reads
bad and stop again once it improves
st_alarm "Get notified" -- send a SmartThings notification when
the air reads bad
(Labels transcribed from the Korean app -- 감지만 하기 / 자동 청정 /
알림받기 -- and rendered here in English; the parenthetical behaviour is
the app's own description of each. The auto-stop half of 'Auto clean' is
not visible in the dump, which reports only the selected autoExeState.)
"""
on = str(rep.get("x.com.samsung.da.periodicSensingActivationState", "")).lower() == "on"
if not on:
return "off"
return {"Airpurify": "auto_purify", "Alarm": "st_alarm"}.get(
str(rep.get("x.com.samsung.da.autoExeState", "")), "sensing_only"
)
def _periodic_sensing_write(payload, rep, href=None):
# The master on/off for periodic sensing. Off holds any auto-action
# pending; On arms the selected one.
return ["airlevelcheck", "vs", "0"], {
"x.com.samsung.da.periodicSensingActivationState": ("On" if payload == "On" else "Off")
}
# One-control version of the same two fields: a single PUT sets the sensing
# toggle and the auto-action together, so 'sensing_only' arms sensing with no
# action in one step instead of two writes that briefly pass through a state
# the user didn't ask for. Confirmed to land both fields in both directions --
# sensing_only -> auto_purify raises autoExeState while activation stays On,
# and back again lowers it.
_SENSING_MODE_BODIES = {
"off": {"x.com.samsung.da.periodicSensingActivationState": "Off"},
"sensing_only": {
"x.com.samsung.da.periodicSensingActivationState": "On",
"x.com.samsung.da.autoExeState": "Off",
},
"auto_purify": {
"x.com.samsung.da.periodicSensingActivationState": "On",
"x.com.samsung.da.autoExeState": "Airpurify",
},
"st_alarm": {
"x.com.samsung.da.periodicSensingActivationState": "On",
"x.com.samsung.da.autoExeState": "Alarm",
},
}
def _sensing_mode_write(payload, rep, href=None):
body = _SENSING_MODE_BODIES.get(payload)
return (["airlevelcheck", "vs", "0"], dict(body)) if body else None
def _skip_status_write(payload, rep, href=None):
return ["airlevelcheck", "vs", "0"], {
"x.com.samsung.da.periodicSensingSkipStatus": ("On" if payload == "On" else "Off")
}
# The daily window during which periodic sensing is skipped, stored as one
# HHMMHHMM string (start+end) on periodicSensingSkipTime. The read side is
# cross-confirmed on two units: issue #84's sits at the inert '00000000', while
# issue #190's carries a real user-set '03002300' -> 03:00-23:00. Split into
# two HA time entities; each write reads the other half back out of the live
# rep so the pair round-trips. Confirmed in both directions on hardware: from
# 13:00-23:00, writing start=07:30 then end=22:00 left the device holding
# '07302200' -- each write kept the half it wasn't given.
def _skip_time_read(part):
def _read(value):
raw = str(value or "")
chunk = raw[0:4] if part == "start" else raw[4:8]
if len(chunk) == 4 and chunk.isdigit():
try:
return datetime.time(int(chunk[:2]), int(chunk[2:]))
except ValueError:
return None
return None
return _read
def _skip_time_write(part):
def _write(value, rep, href=None):
cur = (str(rep.get("x.com.samsung.da.periodicSensingSkipTime", "") or "") + "00000000")[:8]
hhmm = f"{value.hour:02d}{value.minute:02d}"
new = hhmm + cur[4:8] if part == "start" else cur[0:4] + hhmm
return ["airlevelcheck", "vs", "0"], {"x.com.samsung.da.periodicSensingSkipTime": new}
return _write
AIR_LEVEL_CHECK = Capability(
href="/airlevelcheck/vs/0",
poll_tier="warm",
entities=(
SelectDesc(
key="sensing_mode",
rep_fn=_sensing_mode,
options=("off", "sensing_only", "auto_purify", "st_alarm"),
translation_key="sensing_mode",
icon="mdi:radar",
entity_category="config",
write_fn=_sensing_mode_write,
),
SwitchDesc(
key="periodic_air_sensing",
field="x.com.samsung.da.periodicSensingActivationState",
icon="mdi:radar",
entity_category="config",
value_fn=lambda v: str(v).lower() == "on",
write_fn=_periodic_sensing_write,
),
# The one field that varies across the three families reporting this
# resource: the TP1X_DA-AC-AIR dump (#130) omits it while both
# AVT-WW-TP1 dumps and the A-VTWW-TP2 dump carry it, so that board runs
# the sensing engine on a fixed interval it doesn't expose.
NumberDesc(
key="sensing_interval",
field="x.com.samsung.da.periodicSensingInterval",
icon="mdi:timer-cog",
entity_category="config",
native_min=1,
native_max=60,
step=1,
unit="min",
exists_fn=lambda rep, resources: "x.com.samsung.da.periodicSensingInterval" in rep,
value_fn=_interval_minutes,
write_fn=_interval_write,
),
SwitchDesc(
key="periodic_sensing_skip_status",
field="x.com.samsung.da.periodicSensingSkipStatus",
icon="mdi:sleep",
entity_category="config",
value_fn=lambda v: str(v).lower() == "on",
write_fn=_skip_status_write,
),
TimeDesc(
key="sensing_skip_start",
field="x.com.samsung.da.periodicSensingSkipTime",
icon="mdi:clock-start",
entity_category="config",
value_fn=_skip_time_read("start"),
write_fn=_skip_time_write("start"),
),
TimeDesc(
key="sensing_skip_end",
field="x.com.samsung.da.periodicSensingSkipTime",
icon="mdi:clock-end",
entity_category="config",
value_fn=_skip_time_read("end"),
write_fn=_skip_time_write("end"),
),
# Read-only status, same keys as range_hood.AIR_LEVEL_CHECK.
SensorDesc(
key="air_sensing_state",
field="x.com.samsung.da.sensingState",
icon="mdi:radar",
entity_category="diagnostic",
enabled_default=False,
),
SensorDesc(
key="last_air_sensing_time",
field="x.com.samsung.da.lastSensingTime",
device_class="timestamp",
entity_category="diagnostic",
value_fn=epoch_to_utc,
),
# 'Kr1' on both dumps -- a national air-quality grade whose scale is
# region-prefixed and undocumented here, so it stays a raw diagnostic
# rather than being mapped to an asserted enum.
SensorDesc(
key="last_air_sensing_level",
field="x.com.samsung.da.lastSensingLevel",
icon="mdi:air-filter",
entity_category="diagnostic",
enabled_default=False,
),
),
)
# /humidity/0 and /humidity/vs/0 are empty {} on both dumps this family has
# been verified against -- covered here (not globally, per ignored.py's
# module docstring) since those hrefs collide with fridge/AC schemas
@@ -522,7 +765,6 @@ SOUND_VOLUME = Capability(
COVERAGE = [
Capability(href="/humidity/0"),
Capability(href="/humidity/vs/0"),
Capability(href="/airlevelcheck/vs/0"), # periodic air-quality sensing scheduler plumbing
Capability(href="/availablecontrolsets/vs/0"), # opaque hex-encoded control-set bitmap
Capability(href="/da/softreset/vs/0"), # soft-reset trigger plumbing
Capability(href="/keepnormalstate/vs/0"), # internal keep-normal flag
@@ -64,6 +64,18 @@ def parse_iso_utc(raw):
return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC)
def epoch_to_utc(value):
"""Unix epoch seconds -> aware UTC datetime, for the boards that report a
bare epoch rather than the ISO string parse_iso_utc handles. Lived in
range_hood.py as `_timestamp` until air_purifier.py needed the same reading
for /airlevelcheck/vs/0's lastSensingTime -- promoted here rather than
cross-imported, matching how filter_usage_percent was shared."""
try:
return datetime.fromtimestamp(float(value), tz=UTC)
except (TypeError, ValueError, OSError):
return None
def filter_usage_percent(rep):
"""Filter usage as a percentage of rated capacity. Several families
(AC, air purifier) report `filterUsage` as a raw count in
@@ -7,8 +7,6 @@ brightness remain separate controls because the device advertises them as two
independent fields.
"""
from datetime import UTC, datetime
from ..batch import is_stub_rep
from ..capability import Capability
from ..entities import (
@@ -19,14 +17,7 @@ from ..entities import (
SensorDesc,
SwitchDesc,
)
from .common import int_or_none, sensor_item_value
def _timestamp(value):
try:
return datetime.fromtimestamp(float(value), tz=UTC)
except (TypeError, ValueError, OSError):
return None
from .common import epoch_to_utc, int_or_none, sensor_item_value
def _active_alarm_codes(items):
@@ -286,7 +277,7 @@ AIR_LEVEL_CHECK = Capability(
field="x.com.samsung.da.lastSensingTime",
device_class="timestamp",
entity_category="diagnostic",
value_fn=_timestamp,
value_fn=epoch_to_utc,
),
SensorDesc(
key="last_air_sensing_level",
@@ -191,6 +191,9 @@
"oven_setpoint": {
"name": "Požadovaná teplota"
},
"sensing_interval": {
"name": "Interval AI Purify"
},
"setpoint": {
"name": "Požadovaná teplota"
},
@@ -494,6 +497,15 @@
"rinse_cycles": {
"name": "Počet máchání"
},
"sensing_mode": {
"name": "AI Purify",
"state": {
"off": "Vypnuto",
"sensing_only": "Pouze měření",
"auto_purify": "Automatické čištění",
"st_alarm": "Dostat upozornění"
}
},
"softener_concentration": {
"name": "Koncentrace aviváže",
"state": {
@@ -1030,6 +1042,12 @@
"display": {
"name": "Displej"
},
"periodic_air_sensing": {
"name": "Měření AI Purify"
},
"periodic_sensing_skip_status": {
"name": "Přeskočení měření AI Purify"
},
"pet_filter_activation": {
"name": "Aktivace filtru pro domácí mazlíčky"
},
@@ -1199,6 +1217,12 @@
},
"night_start": {
"name": "Začátek nočního osvětlení"
},
"sensing_skip_end": {
"name": "Konec přeskočení měření AI Purify"
},
"sensing_skip_start": {
"name": "Začátek přeskočení měření AI Purify"
}
},
"water_heater": {
@@ -191,6 +191,9 @@
"oven_setpoint": {
"name": "Setpoint"
},
"sensing_interval": {
"name": "AI Purify interval"
},
"setpoint": {
"name": "Setpoint"
},
@@ -494,6 +497,15 @@
"rinse_cycles": {
"name": "Rinse cycles"
},
"sensing_mode": {
"name": "AI Purify",
"state": {
"off": "Off",
"sensing_only": "Sensing only",
"auto_purify": "Auto clean",
"st_alarm": "Get notified"
}
},
"softener_concentration": {
"name": "Softener concentration",
"state": {
@@ -1030,6 +1042,12 @@
"display": {
"name": "Display"
},
"periodic_air_sensing": {
"name": "AI Purify sensing"
},
"periodic_sensing_skip_status": {
"name": "AI Purify sensing skip"
},
"pet_filter_activation": {
"name": "Pet filter activation"
},
@@ -1199,6 +1217,12 @@
},
"night_start": {
"name": "Night light start"
},
"sensing_skip_end": {
"name": "AI Purify sensing skip end"
},
"sensing_skip_start": {
"name": "AI Purify sensing skip start"
}
},
"water_heater": {
@@ -313,6 +313,9 @@
"oven_setpoint": {
"name": "Punto de ajuste"
},
"sensing_interval": {
"name": "Intervalo de AI Purify"
},
"setpoint": {
"name": "Punto de ajuste"
},
@@ -616,6 +619,15 @@
"rinse_cycles": {
"name": "Aclarados"
},
"sensing_mode": {
"name": "AI Purify",
"state": {
"off": "Apagado",
"sensing_only": "Solo medición",
"auto_purify": "Limpieza automática",
"st_alarm": "Recibir notificación"
}
},
"softener_concentration": {
"name": "Concentración del suavizante",
"state": {
@@ -1152,6 +1164,12 @@
"display": {
"name": "Pantalla"
},
"periodic_air_sensing": {
"name": "Medición de AI Purify"
},
"periodic_sensing_skip_status": {
"name": "Omisión de medición de AI Purify"
},
"pet_filter_activation": {
"name": "Activación de filtro para mascotas"
},
@@ -1321,6 +1339,12 @@
},
"night_start": {
"name": "Inicio de la luz nocturna"
},
"sensing_skip_end": {
"name": "Fin de la omisión de medición de AI Purify"
},
"sensing_skip_start": {
"name": "Inicio de la omisión de medición de AI Purify"
}
},
"water_heater": {
@@ -191,6 +191,9 @@
"oven_setpoint": {
"name": "Valore impostato"
},
"sensing_interval": {
"name": "Intervallo AI Purify"
},
"setpoint": {
"name": "Valore impostato"
},
@@ -494,6 +497,15 @@
"rinse_cycles": {
"name": "Risciacquo"
},
"sensing_mode": {
"name": "AI Purify",
"state": {
"off": "Spento",
"sensing_only": "Solo rilevamento",
"auto_purify": "Pulizia automatica",
"st_alarm": "Ricevi notifica"
}
},
"softener_concentration": {
"name": "Concentrazione ammorbidente",
"state": {
@@ -1030,6 +1042,12 @@
"display": {
"name": "Display"
},
"periodic_air_sensing": {
"name": "Rilevamento AI Purify"
},
"periodic_sensing_skip_status": {
"name": "Esclusione rilevamento AI Purify"
},
"pet_filter_activation": {
"name": "Attivazione filtro animali"
},
@@ -1199,6 +1217,12 @@
},
"night_start": {
"name": "Inizio luce notturna"
},
"sensing_skip_end": {
"name": "Fine esclusione rilevamento AI Purify"
},
"sensing_skip_start": {
"name": "Inizio esclusione rilevamento AI Purify"
}
},
"water_heater": {
@@ -191,6 +191,9 @@
"oven_setpoint": {
"name": "Instelpunt"
},
"sensing_interval": {
"name": "AI Purify-interval"
},
"setpoint": {
"name": "Instelpunt"
},
@@ -494,6 +497,15 @@
"rinse_cycles": {
"name": "Spoelcycli"
},
"sensing_mode": {
"name": "AI Purify",
"state": {
"off": "Uit",
"sensing_only": "Alleen meten",
"auto_purify": "Automatisch reinigen",
"st_alarm": "Melding ontvangen"
}
},
"softener_concentration": {
"name": "Concentratie wasverzachter",
"state": {
@@ -1030,6 +1042,12 @@
"display": {
"name": "Display"
},
"periodic_air_sensing": {
"name": "AI Purify-meting"
},
"periodic_sensing_skip_status": {
"name": "AI Purify meting overslaan"
},
"pet_filter_activation": {
"name": "Huisdierfilter actief"
},
@@ -1199,6 +1217,12 @@
},
"night_start": {
"name": "Start nachtverlichting"
},
"sensing_skip_end": {
"name": "Einde AI Purify meting overslaan"
},
"sensing_skip_start": {
"name": "Start AI Purify meting overslaan"
}
},
"water_heater": {
+9
View File
@@ -1,5 +1,6 @@
{
"state_keys": [
"air_sensing_state",
"alarm_code",
"clean_level",
"device_active",
@@ -10,9 +11,17 @@
"firmware_update",
"hepa_filter_status",
"hepa_filter_usage",
"last_air_sensing_level",
"last_air_sensing_time",
"mute_once",
"odor",
"periodic_air_sensing",
"periodic_sensing_skip_status",
"power_switch",
"sensing_interval",
"sensing_mode",
"sensing_skip_end",
"sensing_skip_start",
"super_fine_dust",
"wind_strength_fan"
]
@@ -1,5 +1,6 @@
{
"state_keys": [
"air_sensing_state",
"alarm_code",
"child_lock",
"clean_level",
@@ -14,11 +15,18 @@
"firmware_update",
"hepa_filter_status",
"hepa_filter_usage",
"last_air_sensing_level",
"last_air_sensing_time",
"mute_once",
"odor",
"panel_status",
"periodic_air_sensing",
"periodic_sensing_skip_status",
"pet_filter_activation",
"power_switch",
"sensing_mode",
"sensing_skip_end",
"sensing_skip_start",
"sound_mode",
"sound_output",
"sound_volume",
+9
View File
@@ -1,5 +1,6 @@
{
"state_keys": [
"air_sensing_state",
"alarm_code",
"clean_level",
"device_active",
@@ -9,8 +10,16 @@
"firmware_update",
"hepa_filter_status",
"hepa_filter_usage",
"last_air_sensing_level",
"last_air_sensing_time",
"odor",
"periodic_air_sensing",
"periodic_sensing_skip_status",
"power_switch",
"sensing_interval",
"sensing_mode",
"sensing_skip_end",
"sensing_skip_start",
"super_fine_dust",
"wind_strength_fan"
]
+224
View File
@@ -0,0 +1,224 @@
"""Tests for /airlevelcheck/vs/0 -- the "AI Purify" periodic air-quality
sensing engine (issues #84 and #190).
The resource is reported by three of this registry's four board families, so
the read assertions run against each family's own fixture; the write contracts
were exercised on AVT-WW-TP1-23-AXX500 hardware and are asserted here at the
body level.
"""
import datetime
from custom_components.localthings.registry.adapter import flatten
from custom_components.localthings.registry.by_type import resolve
from custom_components.localthings.registry.capabilities import air_purifier
from custom_components.localthings.registry.discovery import discover
from tests.conftest import _load_device
# The three fixtures whose dumps carry this resource. air_purifier (the
# ARTIK051_TVTL family, issue #56) has no such href and is deliberately absent.
FAMILIES = ("air_purifier_avt_ww", "air_purifier_vtww", "air_purifier_tp1x_da_ac_air")
def _state(fixture):
resources = _load_device(fixture)
reg = resolve(resources)
assert reg is not None and reg.name == "air_purifier", fixture
return flatten(discover(resources, reg.capabilities, reg.pattern_capabilities), resources)
def _desc(key):
return next(d for d in air_purifier.AIR_LEVEL_CHECK.entities if d.key == key)
def test_air_level_check_is_bound_not_covered():
"""The href used to sit in COVERAGE as opaque scheduler plumbing. Guard
against it being covered again, which would silently drop every entity
below while still reporting zero unbound hrefs."""
covered = {cap.href for cap in air_purifier.COVERAGE}
assert "/airlevelcheck/vs/0" not in covered
def test_every_reporting_family_binds_the_cluster():
for fixture in FAMILIES:
state = _state(fixture)
for key in (
"sensing_mode",
"periodic_air_sensing",
"periodic_sensing_skip_status",
"sensing_skip_start",
"sensing_skip_end",
"air_sensing_state",
"last_air_sensing_time",
"last_air_sensing_level",
):
assert key in state, f"{fixture}: {key}"
def test_tvtl_family_is_untouched():
"""Issue #56's board has no /airlevelcheck href at all -- nothing this
change adds may appear on it."""
state = _state("air_purifier")
for key in ("sensing_mode", "periodic_air_sensing", "sensing_interval", "sensing_skip_start"):
assert key not in state, key
def test_sensing_interval_only_where_the_field_exists():
"""TP1X_DA-AC-AIR (issue #130) omits periodicSensingInterval; the other two
report it. The entity must follow the field, not the href."""
assert "sensing_interval" in _state("air_purifier_avt_ww")
assert "sensing_interval" in _state("air_purifier_vtww")
assert "sensing_interval" not in _state("air_purifier_tp1x_da_ac_air")
def test_no_unbound_hrefs_on_any_reporting_family():
for fixture in FAMILIES:
resources = _load_device(fixture)
reg = resolve(resources)
assert reg is not None, fixture
unbound = []
discover(resources, reg.capabilities, reg.pattern_capabilities, log=unbound.append)
assert unbound == [], f"{fixture}: {unbound}"
def test_sensing_mode_folds_toggle_and_action():
"""Off beats any pending auto-action; On with no action is 'sensing only'."""
assert (
air_purifier._sensing_mode(
{
"x.com.samsung.da.periodicSensingActivationState": "Off",
"x.com.samsung.da.autoExeState": "Airpurify",
}
)
== "off"
)
assert (
air_purifier._sensing_mode(
{
"x.com.samsung.da.periodicSensingActivationState": "On",
"x.com.samsung.da.autoExeState": "Off",
}
)
== "sensing_only"
)
assert (
air_purifier._sensing_mode(
{
"x.com.samsung.da.periodicSensingActivationState": "On",
"x.com.samsung.da.autoExeState": "Airpurify",
}
)
== "auto_purify"
)
assert (
air_purifier._sensing_mode(
{
"x.com.samsung.da.periodicSensingActivationState": "On",
"x.com.samsung.da.autoExeState": "Alarm",
}
)
== "st_alarm"
)
def test_sensing_mode_write_sets_both_fields_in_one_body():
href, body = air_purifier._sensing_mode_write("auto_purify", {})
assert href == ["airlevelcheck", "vs", "0"]
assert body == {
"x.com.samsung.da.periodicSensingActivationState": "On",
"x.com.samsung.da.autoExeState": "Airpurify",
}
# 'off' only needs the toggle -- the pending action is preserved.
assert air_purifier._sensing_mode_write("off", {})[1] == {
"x.com.samsung.da.periodicSensingActivationState": "Off"
}
assert air_purifier._sensing_mode_write("nonsense", {}) is None
def test_sensing_mode_write_does_not_mutate_the_shared_body_table():
before = dict(air_purifier._SENSING_MODE_BODIES["auto_purify"])
_, body = air_purifier._sensing_mode_write("auto_purify", {})
body["x.com.samsung.da.autoExeState"] = "clobbered"
assert air_purifier._SENSING_MODE_BODIES["auto_purify"] == before
def test_interval_is_minutes_in_the_ui_and_seconds_on_the_wire():
assert air_purifier._interval_minutes("600") == 10
assert air_purifier._interval_minutes(None) is None
assert air_purifier._interval_write(10, {})[1] == {
"x.com.samsung.da.periodicSensingInterval": "600"
}
def test_skip_time_splits_the_hhmmhhmm_window():
read_start = air_purifier._skip_time_read("start")
read_end = air_purifier._skip_time_read("end")
# Issue #190's unit ships a real window: 03:00-23:00.
assert read_start("03002300") == datetime.time(3, 0)
assert read_end("03002300") == datetime.time(23, 0)
# Issue #84's unit sits at the inert default.
assert read_start("00000000") == datetime.time(0, 0)
# Junk and short strings read as unknown rather than raising.
assert read_start("") is None
assert read_start("99999999") is None
assert read_end("0300") is None
def test_skip_time_write_preserves_the_other_half():
rep = {"x.com.samsung.da.periodicSensingSkipTime": "03002300"}
_, body = air_purifier._skip_time_write("start")(datetime.time(7, 30), rep)
assert body == {"x.com.samsung.da.periodicSensingSkipTime": "07302300"}
_, body = air_purifier._skip_time_write("end")(datetime.time(22, 5), rep)
assert body == {"x.com.samsung.da.periodicSensingSkipTime": "03002205"}
# A board that has never had a window set still round-trips.
_, body = air_purifier._skip_time_write("end")(datetime.time(1, 2), {})
assert body == {"x.com.samsung.da.periodicSensingSkipTime": "00000102"}
def test_periodic_sensing_and_skip_switch_bodies():
assert air_purifier._periodic_sensing_write("On", {})[1] == {
"x.com.samsung.da.periodicSensingActivationState": "On"
}
assert air_purifier._periodic_sensing_write("Off", {})[1] == {
"x.com.samsung.da.periodicSensingActivationState": "Off"
}
assert air_purifier._skip_status_write("On", {})[1] == {
"x.com.samsung.da.periodicSensingSkipStatus": "On"
}
def test_last_sensing_time_reads_as_utc():
state = _state("air_purifier_avt_ww")
assert state["last_air_sensing_time"].tzinfo is not None
assert state["last_air_sensing_time"].year >= 2020
def test_read_only_keys_match_the_range_hood_capability():
"""Same href, same fields -- the keys are shared deliberately so both
families read from one translation catalog entry. If either side renames
one, this catches the drift."""
from custom_components.localthings.registry.capabilities import range_hood
hood = {d.key for d in range_hood.AIR_LEVEL_CHECK.entities}
ours = {d.key for d in air_purifier.AIR_LEVEL_CHECK.entities}
assert {
"air_sensing_state",
"last_air_sensing_time",
"last_air_sensing_level",
"periodic_air_sensing",
} <= hood & ours
def test_periodic_air_sensing_is_writable_here_and_read_only_on_hoods():
"""The reason range_hood.AIR_LEVEL_CHECK is not imported directly: the hood
models this key as a read-only BinarySensorDesc, this board needs a
writable SwitchDesc. Reusing the hood's capability would migrate every hood
user's entity to a different platform."""
from custom_components.localthings.registry.capabilities import range_hood
from custom_components.localthings.registry.entities import BinarySensorDesc, SwitchDesc
hood = next(d for d in range_hood.AIR_LEVEL_CHECK.entities if d.key == "periodic_air_sensing")
assert isinstance(hood, BinarySensorDesc)
ours = _desc("periodic_air_sensing")
assert isinstance(ours, SwitchDesc)
assert ours.write_fn is not None