bhetherman/localthings
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Apply ruff format to code merged from PR #251/#275

Browse Source 
PR #251 and PR #275 predate this repo's ruff-format adoption on those
files; running the formatter (single->double quotes, line wrapping,
trailing-comma cleanup) keeps the merged code consistent with the rest
of the codebase. No behavior change.
This commit is contained in:
Marc Billow
2026-08-08 21:24:02 +00:00
parent 68dee12eb4
commit 8ed3d1467f
10 changed files with 222 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/localthings/coordinator.py
+5 -10
View File
@@ -658,7 +658,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
# ------------------------------------------------------------------
def _subdevice_probe_priority(
self, resources: dict[str, dict],
self,
resources: dict[str, dict],
) -> tuple[str, ...]:
"""Return live primary-entity hrefs in hot/warm-first order.
@@ -676,23 +677,17 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
if registry is None:
return ()
tier_rank = {'hot': 0, 'warm': 1, 'cold': 2}
tier_rank = {"hot": 0, "warm": 1, "cold": 2}
ranked = []
for order, href in enumerate(resources):
primary_caps = [
capability
for capability in registry.capabilities.get(href, ())
if any(
desc.entity_category is None
for desc in capability.entities
)
if any(desc.entity_category is None for desc in capability.entities)
]
if not primary_caps:
continue
rank = min(
tier_rank.get(capability.poll_tier, 2)
for capability in primary_caps
)
rank = min(tier_rank.get(capability.poll_tier, 2) for capability in primary_caps)
ranked.append((rank, order, href))
return tuple(href for _, _, href in sorted(ranked))
custom_components/localthings/registry/capabilities/laundry.py
+7 -7
View File
@@ -20,9 +20,9 @@ Door-LED keys use NO `x.com.samsung.da.` prefix -- `setBrightness` /
`setNightLight` -- preserved exactly as they appear in the OCF resource rep.
"""
import string
from datetime import UTC, datetime
from datetime import time as dt_time
import string
from ...catalog import has_entity_translation
from ..capability import Capability
@@ -324,7 +324,7 @@ def cycle_write(p, rep, href=None):
}
def personal_course_labels(resources, href='/wm/personalcourse/vs/0'):
def personal_course_labels(resources, href="/wm/personalcourse/vs/0"):
"""Return device-provided personal course names keyed by course code.
Populated entries use a small TLV payload. The leading field is
@@ -335,10 +335,10 @@ def personal_course_labels(resources, href='/wm/personalcourse/vs/0'):
"""
rep = resources.get(href) or {}
labels = {}
for entry in rep.get('x.com.samsung.da.courses') or []:
if not isinstance(entry, str) or '_' not in entry:
for entry in rep.get("x.com.samsung.da.courses") or []:
if not isinstance(entry, str) or "_" not in entry:
continue
code, encoded = entry.split('_', 1)
code, encoded = entry.split("_", 1)
try:
payload = bytes.fromhex(encoded)
except ValueError:
@@ -349,7 +349,7 @@ def personal_course_labels(resources, href='/wm/personalcourse/vs/0'):
if name_length == 0 or len(payload) < 2 + name_length:
continue
try:
name = payload[2:2 + name_length].decode('utf-8')
name = payload[2 : 2 + name_length].decode("utf-8")
except UnicodeDecodeError:
continue
if name.strip() and name.isprintable():
@@ -364,7 +364,7 @@ def washer_cycle_fallback(value, resources):
if label := personal_course_labels(resources).get(value.upper()):
return label
if len(value) == 2 and all(char in string.hexdigits for char in value):
return f'Unknown (0x{value.upper()})'
return f"Unknown (0x{value.upper()})"
return None
custom_components/localthings/registry/subdevices.py
+7 -6
View File
@@ -236,7 +236,9 @@ def _get_raw(sess, path_segs: tuple[str, ...], timeout: float = 10.0):
def _get_batch(
sess, path_segs: tuple[str, ...], timeout: float = 10.0,
sess,
path_segs: tuple[str, ...],
timeout: float = 10.0,
) -> 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,
@@ -246,7 +248,9 @@ def _get_batch(
def _get_property(
sess, path_segs: tuple[str, ...], timeout: float = 10.0,
sess,
path_segs: tuple[str, ...],
timeout: float = 10.0,
) -> 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
@@ -301,10 +305,7 @@ def enumerate_subdevices(
# casing, and probing it twice would materialize the same physical
# subdevice as two Subdevice candidates.
probed_ids: set[str] = set()
deadline = (
time.monotonic() + max(0.0, time_budget)
if time_budget is not None else None
)
deadline = time.monotonic() + max(0.0, time_budget) if time_budget is not None else None
budget_exhausted = False
def _next_timeout(maximum: float) -> float | None:
custom_components/localthings/select.py
+1 -2
View File
@@ -99,8 +99,7 @@ class LocalThingsSelect(LocalThingsEntity, SelectEntity):
"""Normalize both current state and options through one path."""
display_fn = self._bound.desc.display_fn
fallback_fn = (
(lambda raw: display_fn(raw, self._resources))
if display_fn is not None else None
(lambda raw: display_fn(raw, self._resources)) if display_fn is not None else None
)
return _display(value, self.translation_key, fallback_fn)
tests/test_laundry_capabilities.py
+21 -21
View File
@@ -67,28 +67,28 @@ class TestCourseHelpers:
assigning an inferred meaning to any standard course code.
"""
resources = {
'/wm/personalcourse/vs/0': {
'x.com.samsung.da.courses': [
'F1_0106EC868DEC98B7021EED8CACED8BB020EB93B1',
'F2_00',
'F3_0109EC9A94EAB8B0EBB3B40220ECA084EC9AA9',
"/wm/personalcourse/vs/0": {
"x.com.samsung.da.courses": [
"F1_0106EC868DEC98B7021EED8CACED8BB020EB93B1",
"F2_00",
"F3_0109EC9A94EAB8B0EBB3B40220ECA084EC9AA9",
],
},
}
first_name = bytes.fromhex('EC868DEC98B7').decode('utf-8')
second_name = bytes.fromhex('EC9A94EAB8B0EBB3B4').decode('utf-8')
first_name = bytes.fromhex("EC868DEC98B7").decode("utf-8")
second_name = bytes.fromhex("EC9A94EAB8B0EBB3B4").decode("utf-8")
assert laundry.personal_course_labels(resources) == {
'F1': first_name,
'F3': second_name,
"F1": first_name,
"F3": second_name,
}
def test_personal_course_names_reject_malformed_payloads(self):
resources = {
'/wm/personalcourse/vs/0': {
'x.com.samsung.da.courses': [
'F1_not-hex',
'F2_0106AA',
'F3_020141',
"/wm/personalcourse/vs/0": {
"x.com.samsung.da.courses": [
"F1_not-hex",
"F2_0106AA",
"F3_020141",
None,
],
},
@@ -97,15 +97,15 @@ class TestCourseHelpers:
def test_washer_cycle_fallback_uses_labels_then_safe_unknown_text(self):
resources = {
'/wm/personalcourse/vs/0': {
'x.com.samsung.da.courses': ['F1_0106EC868DEC98B7'],
"/wm/personalcourse/vs/0": {
"x.com.samsung.da.courses": ["F1_0106EC868DEC98B7"],
},
}
expected = bytes.fromhex('EC868DEC98B7').decode('utf-8')
assert laundry.washer_cycle_fallback('F1', resources) == expected
assert laundry.washer_cycle_fallback('69', resources) == 'Unknown (0x69)'
assert laundry.washer_cycle_fallback('6f', resources) == 'Unknown (0x6F)'
assert laundry.washer_cycle_fallback('Normal', resources) is None
expected = bytes.fromhex("EC868DEC98B7").decode("utf-8")
assert laundry.washer_cycle_fallback("F1", resources) == expected
assert laundry.washer_cycle_fallback("69", resources) == "Unknown (0x69)"
assert laundry.washer_cycle_fallback("6f", resources) == "Unknown (0x6F)"
assert laundry.washer_cycle_fallback("Normal", resources) is None
class TestCourseCodesFromSupportedOptions:
tests/test_select_display.py
+1 -3
View File
@@ -49,6 +49,4 @@ def test_display_passes_through_non_string_values():
def test_display_uses_fallback_when_translation_has_no_state_table():
assert _display('69', 'cycle', lambda value: f'Unknown (0x{value})') == (
'Unknown (0x69)'
)
assert _display("69", "cycle", lambda value: f"Unknown (0x{value})") == ("Unknown (0x69)")
tests/test_subdevice_discovery.py
+1
View File
@@ -277,6 +277,7 @@ async def test_fac_bora_2in1_unique_ids_include_subdevice_prefix(hass: HomeAssis
# capture instead of the synthetic sessions test_subdevices.py uses.
# ---------------------------------------------------------------------------
async def test_flat_probe_priority_puts_live_climate_state_before_cold_metrics(
hass: HomeAssistant,
):
tests/test_subdevices.py
+22 -20
View File
@@ -360,6 +360,7 @@ def test_enumeration_budget_bounds_silent_prefixed_fallback_and_uses_priority(
live-state href is then attempted first, and the last Property probe is
clamped to exactly the time left in the shared enumeration budget.
"""
class Clock:
now = 0.0
@@ -381,20 +382,20 @@ def test_enumeration_budget_bounds_silent_prefixed_fallback_and_uses_priority(
clock = Clock()
session = SilentSession(clock)
monkeypatch.setattr(subdevices_module.time, 'monotonic', clock.monotonic)
monkeypatch.setattr(subdevices_module.time, "monotonic", clock.monotonic)
resources = {
'/subdevices/vs/0': {
'x.com.samsung.da.subdeviceIdList': [_UUID],
"/subdevices/vs/0": {
"x.com.samsung.da.subdeviceIdList": [_UUID],
},
'/power/vs/0': {'power': 'On'},
'/mode/vs/0': {'mode': 'Cool'},
"/power/vs/0": {"power": "On"},
"/mode/vs/0": {"mode": "Cool"},
}
found, extra = enumerate_subdevices(
session,
resources,
oic_res_links=[],
preferred_hrefs=('/mode/vs/0',),
preferred_hrefs=("/mode/vs/0",),
time_budget=7.0,
collection_timeout=4.0,
property_timeout=2.0,
@@ -404,9 +405,9 @@ def test_enumeration_budget_bounds_silent_prefixed_fallback_and_uses_priority(
assert extra == {}
assert clock.now == 7.0
assert session.calls == [
((_UUID, 'device', '0'), 4.0),
((_UUID, 'mode', 'vs', '0'), 2.0),
((_UUID, 'power', 'vs', '0'), 1.0),
((_UUID, "device", "0"), 4.0),
((_UUID, "mode", "vs", "0"), 2.0),
((_UUID, "power", "vs", "0"), 1.0),
]
@@ -414,6 +415,7 @@ def test_enumeration_keeps_preferred_response_found_before_budget_expires(
monkeypatch,
):
"""A useful early response survives later silent probes hitting the cap."""
class Clock:
now = 0.0
@@ -425,8 +427,8 @@ def test_enumeration_keeps_preferred_response_found_before_budget_expires(
self.clock = clock
def get(self, path, timeout=10.0):
if tuple(path) == (_UUID, 'mode', 'vs', '0'):
return 0x45, cbor2.dumps({'mode': 'Cool'})
if tuple(path) == (_UUID, "mode", "vs", "0"):
return 0x45, cbor2.dumps({"mode": "Cool"})
self.clock.now += timeout
raise TimeoutError
@@ -434,30 +436,30 @@ def test_enumeration_keeps_preferred_response_found_before_budget_expires(
pass
clock = Clock()
monkeypatch.setattr(subdevices_module.time, 'monotonic', clock.monotonic)
monkeypatch.setattr(subdevices_module.time, "monotonic", clock.monotonic)
resources = {
'/subdevices/vs/0': {
'x.com.samsung.da.subdeviceIdList': [_UUID],
"/subdevices/vs/0": {
"x.com.samsung.da.subdeviceIdList": [_UUID],
},
'/power/vs/0': {'power': 'On'},
'/mode/vs/0': {'mode': 'Cool'},
"/power/vs/0": {"power": "On"},
"/mode/vs/0": {"mode": "Cool"},
}
found, extra = enumerate_subdevices(
PartlyResponsiveSession(clock),
resources,
oic_res_links=[],
preferred_hrefs=('/mode/vs/0',),
preferred_hrefs=("/mode/vs/0",),
time_budget=7.0,
collection_timeout=4.0,
property_timeout=2.0,
)
assert [(subdevice.kind, subdevice.key) for subdevice in found] == [
('prefixed', _UUID),
("prefixed", _UUID),
]
assert found[0].flat_hrefs == ('/mode/vs/0',)
assert extra == {f'/{_UUID}/mode/vs/0': {'mode': 'Cool'}}
assert found[0].flat_hrefs == ("/mode/vs/0",)
assert extra == {f"/{_UUID}/mode/vs/0": {"mode": "Cool"}}
assert clock.now == 7.0
tests/test_translations.py
+138 -64
View File
@@ -115,90 +115,164 @@ def test_no_catalog_carries_unresolved_core_references():
def test_confirmed_korean_table_02_washer_course_names():
states = _load('en')['entity']['select']['washer_cycle_table_02']['state']
assert {code: states[code] for code in (
'69', '6a', '6b', '6c', '6d', '6e', '6f', '70', '71',
'72', '73', '74', '75', '76', '77', '78', '79', '88',
)} == {
'69': 'AI Wash',
'6a': 'Wool',
'6b': 'Denim',
'6c': 'Blouses',
'6d': 'Delicates',
'6e': 'Active Wear',
'6f': 'Bedding',
'70': 'Towels',
'71': 'Quick Wash',
'72': 'Shirts',
'73': 'Sanitize',
'74': 'Drum Clean',
'75': 'Outdoor',
'76': 'Baby Care',
'77': 'Cottons',
'78': 'Rinse + Spin',
'79': 'Spin Only',
'88': 'Pet Care',
states = _load("en")["entity"]["select"]["washer_cycle_table_02"]["state"]
assert {
code: states[code]
for code in (
"69",
"6a",
"6b",
"6c",
"6d",
"6e",
"6f",
"70",
"71",
"72",
"73",
"74",
"75",
"76",
"77",
"78",
"79",
"88",
)
} == {
"69": "AI Wash",
"6a": "Wool",
"6b": "Denim",
"6c": "Blouses",
"6d": "Delicates",
"6e": "Active Wear",
"6f": "Bedding",
"70": "Towels",
"71": "Quick Wash",
"72": "Shirts",
"73": "Sanitize",
"74": "Drum Clean",
"75": "Outdoor",
"76": "Baby Care",
"77": "Cottons",
"78": "Rinse + Spin",
"79": "Spin Only",
"88": "Pet Care",
}
def test_confirmed_dishwasher_course_names():
states = _load('en')['entity']['select']['dishwasher_cycle']['state']
assert {code: states[code] for code in (
'82', '8a', 'a7', 'a8', '8c', '88',
)} == {
'82': 'Auto',
'8a': 'Normal',
'a7': 'Heavy',
'a8': 'Express',
'8c': 'Extra Silence',
'88': 'Self Clean',
states = _load("en")["entity"]["select"]["dishwasher_cycle"]["state"]
assert {
code: states[code]
for code in (
"82",
"8a",
"a7",
"a8",
"8c",
"88",
)
} == {
"82": "Auto",
"8a": "Normal",
"a7": "Heavy",
"a8": "Express",
"8c": "Extra Silence",
"88": "Self Clean",
}
def test_confirmed_course_names_are_localized():
washer_codes = (
'69', '6a', '6b', '6c', '6d', '6e', '6f', '70', '71',
'72', '73', '74', '75', '76', '77', '78', '79', '88',
"69",
"6a",
"6b",
"6c",
"6d",
"6e",
"6f",
"70",
"71",
"72",
"73",
"74",
"75",
"76",
"77",
"78",
"79",
"88",
)
dishwasher_codes = ('82', '8a', 'a7', 'a8', '8c', '88')
dishwasher_codes = ("82", "8a", "a7", "a8", "8c", "88")
expected = {
'cs': {
'washer': (
'AI praní', 'Vlna', 'Džíny', 'Halenky', 'Jemné prádlo',
'Sportovní oblečení', 'Ložní prádlo', 'Ručníky',
'Rychlé praní', 'Košile', 'Dezinfekce', 'Čištění bubnu',
'Outdoor', 'Dětské potřeby', 'Bavlna',
'Máchání + odstřeďování', 'Pouze odstřeďování',
'Péče o domácí mazlíčky',
"cs": {
"washer": (
"AI praní",
"Vlna",
"Džíny",
"Halenky",
"Jemné prádlo",
"Sportovní oblečení",
"Ložní prádlo",
"Ručníky",
"Rychlé praní",
"Košile",
"Dezinfekce",
"Čištění bubnu",
"Outdoor",
"Dětské potřeby",
"Bavlna",
"Máchání + odstřeďování",
"Pouze odstřeďování",
"Péče o domácí mazlíčky",
),
'dishwasher': (
'Automatický', 'Normální', 'Intenzivní', 'Expresní',
'Extra tichý', 'Samočištění',
"dishwasher": (
"Automatický",
"Normální",
"Intenzivní",
"Expresní",
"Extra tichý",
"Samočištění",
),
},
'nl': {
'washer': (
'AI Wash', 'Wol', 'Spijkergoed', 'Blouses', 'Fijne was',
'Sportkleding', 'Beddengoed', 'Handdoeken', 'Snelle was',
'Overhemden', 'Hygiëne', 'Trommel reinigen', 'Outdoor',
'Babyverzorging', 'Katoen', 'Spoelen + centrifugeren',
'Alleen centrifugeren', 'Pet Care',
"nl": {
"washer": (
"AI Wash",
"Wol",
"Spijkergoed",
"Blouses",
"Fijne was",
"Sportkleding",
"Beddengoed",
"Handdoeken",
"Snelle was",
"Overhemden",
"Hygiëne",
"Trommel reinigen",
"Outdoor",
"Babyverzorging",
"Katoen",
"Spoelen + centrifugeren",
"Alleen centrifugeren",
"Pet Care",
),
'dishwasher': (
'Auto', 'Normaal', 'Intensief', 'Express', 'Extra stil',
'Zelfreiniging',
"dishwasher": (
"Auto",
"Normaal",
"Intensief",
"Express",
"Extra stil",
"Zelfreiniging",
),
},
}
for language, translations in expected.items():
catalog = _load(language)['entity']['select']
washer = catalog['washer_cycle_table_02']['state']
dishwasher = catalog['dishwasher_cycle']['state']
assert tuple(washer[code] for code in washer_codes) == translations['washer']
assert tuple(dishwasher[code] for code in dishwasher_codes) == (
translations['dishwasher']
)
catalog = _load(language)["entity"]["select"]
washer = catalog["washer_cycle_table_02"]["state"]
dishwasher = catalog["dishwasher_cycle"]["state"]
assert tuple(washer[code] for code in washer_codes) == translations["washer"]
assert tuple(dishwasher[code] for code in dishwasher_codes) == (translations["dishwasher"])
# The hood fan is its device's primary feature: fan.py sets _attr_name = None
tests/test_washer_capabilities.py
+19 -3
View File
@@ -125,10 +125,26 @@ class TestWasherCourse:
from custom_components.localthings.catalog import translated_states
confirmed = {
'69', '6a', '6b', '6c', '6d', '6e', '6f', '70', '71',
'72', '73', '74', '75', '76', '77', '78', '79', '88',
"69",
"6a",
"6b",
"6c",
"6d",
"6e",
"6f",
"70",
"71",
"72",
"73",
"74",
"75",
"76",
"77",
"78",
"79",
"88",
}
assert confirmed <= translated_states('select', 'washer_cycle_table_02')
assert confirmed <= translated_states("select", "washer_cycle_table_02")
def test_missing_course_option_returns_none(self):
desc = next(e for e in washer.WASHER_COURSE.entities if e.key == "cycle")