feat: replace MQTT bridge with localthings HA custom component

Convert from MQTT bridge architecture to a proper Home Assistant
integration (custom_components/localthings/):

- Config flow: IP-only entry, probes ports [49154,49155], cert/key
  reused from first device on subsequent adds
- Coordinator: DataUpdateCoordinator wrapping blocking DTLS session
  via async_add_executor_job; discovery on first poll; dynamic 5s/30s
  interval based on active_when
- Entity platforms: sensor, binary_sensor, switch, number, select
- Library cleanup: delete bridge.py, main.py, config.py, logger.py,
  keepalive.py, poll_scheduler.py, sensors.py, state_cache.py
- adapter.py stripped to flatten(), is_active(), _key()
- _parse_device0_batch moved to samsung_appliance/batch.py

tests/conftest.py

@@ -7,8 +7,8 @@ FIXTURES = Path(__file__).resolve().parent / 'fixtures'
 
 
 def _resources_from_dump(dump: dict) -> dict[str, dict]:
-    from samsung_appliance.bridge import _parse_device0_batch
-    return _parse_device0_batch(dump['device0'])
+    from samsung_appliance.batch import parse_device0_batch
+    return parse_device0_batch(dump['device0'])
 
 
 def _load_device(name: str) -> dict[str, dict]:

tests/test_adapter.py

@@ -1,88 +0,0 @@
-import json
-
-from samsung_appliance.registry.adapter import build_runtime_descriptor
-from samsung_appliance.registry.capabilities import common
-from samsung_appliance.registry.discovery import discover
-
-
-def _bound(resources):
-    reg = {c.href: [c] for c in (common.KIDS_LOCK, common.ENERGY_METER)}
-    return discover(resources, reg)
-
-
-def _bound_with_power(resources):
-    reg = {c.href: [c] for c in (common.KIDS_LOCK, common.ENERGY_METER, common.POWER)}
-    return discover(resources, reg)
-
-
-def test_flatten_applies_value_fn(dishwasher_resources):
-    rd = build_runtime_descriptor(
-        _bound(dishwasher_resources), topic_prefix='t', ha_prefix='homeassistant',
-        device_name='Dishwasher', model='M', name='dishwasher', default_port=49154)
-    flat = rd.flatten(dishwasher_resources)
-    assert flat['power_watts'] is not None
-    assert isinstance(flat['child_lock'], bool)
-
-
-def test_discovery_payloads_have_unique_ids(dishwasher_resources):
-    rd = build_runtime_descriptor(
-        _bound(dishwasher_resources), topic_prefix='samsung_dishwasher',
-        ha_prefix='homeassistant', device_name='Dishwasher', model='M',
-        name='dishwasher', default_port=49154)
-    topics = [t for t, _ in rd.discovery_payloads]
-    # one config topic per produced sensor/binary_sensor entity
-    assert any('/binary_sensor/samsung_dishwasher/child_lock/config' in t for t in topics)
-    assert any('/sensor/samsung_dishwasher/power_watts/config' in t for t in topics)
-    # payloads are JSON bytes carrying unique_id
-    cfg = json.loads(next(p for t, p in rd.discovery_payloads if 'child_lock' in t))
-    assert cfg['unique_id'] == 'samsung_dishwasher_child_lock'
-
-
-def test_switch_discovery_config_has_correct_payloads(dishwasher_resources):
-    rd = build_runtime_descriptor(
-        _bound_with_power(dishwasher_resources), topic_prefix='samsung_dishwasher',
-        ha_prefix='homeassistant', device_name='Dishwasher', model='M',
-        name='dishwasher', default_port=49154)
-    cfg = json.loads(next(p for t, p in rd.discovery_payloads if 'power_switch' in t))
-    assert cfg['payload_on'] == 'On'
-    assert cfg['payload_off'] == 'Off'
-    assert cfg['state_on'] == 'On'
-    assert cfg['state_off'] == 'Off'
-    assert 'On' in cfg['value_template']
-    assert 'Off' in cfg['value_template']
-
-
-def test_bound_entity_key_override_takes_precedence():
-    from samsung_appliance.registry.adapter import _key
-    from samsung_appliance.registry.discovery import BoundEntity
-    from samsung_appliance.registry.entities import SensorDesc
-
-    cap = common.ENERGY_METER
-    desc = SensorDesc(
-        key='power_watts',
-        field='x.com.samsung.da.power',
-        unit='W',
-        device_class='power',
-        state_class='measurement',
-        value_fn=lambda x: x,
-    )
-
-    # Without key_override, should use desc.key
-    b1 = BoundEntity(href='/power/vs/0', capability=cap, desc=desc, instance='')
-    assert _key(b1) == 'power_watts'
-
-    # With key_override, should use key_override
-    b2 = BoundEntity(href='/power/vs/0', capability=cap, desc=desc, instance='', key_override='foo')
-    assert _key(b2) == 'foo'
-
-    # key_override with instance suffix should preserve instance suffix
-    b3 = BoundEntity(href='/power/vs/0', capability=cap, desc=desc, instance='_1', key_override='bar')
-    assert _key(b3) == 'bar_1'
-
-
-def test_runtime_descriptor_has_active_and_idle_intervals(dishwasher_resources):
-    rd = build_runtime_descriptor(
-        _bound(dishwasher_resources), topic_prefix='t', ha_prefix='homeassistant',
-        device_name='Dishwasher', model='M', name='dishwasher', default_port=49154)
-    assert rd.active_interval_s == 5.0
-    assert rd.idle_interval_s == 30.0

tests/test_end_to_end.py

@@ -1,24 +0,0 @@
-import pytest
-
-from tests.conftest import _load_resources
-from samsung_appliance.registry import (CAPABILITIES, discover,
-                                        build_runtime_descriptor)
-
-
-@pytest.mark.parametrize('name,ip', [
-    ('dishwasher', '10.0.0.129'),
-    ('refrigerator', '10.0.0.254'),
-])
-def test_full_pipeline_produces_entities_and_intervals(name, ip):
-    resources = _load_resources(ip)
-    bound = discover(resources, CAPABILITIES)
-    rd = build_runtime_descriptor(
-        bound, topic_prefix=f'samsung_{name}', ha_prefix='homeassistant',
-        device_name=name.title(), model='M', name=name, default_port=49154)
-    assert rd.discovery_payloads, "no entities discovered"
-    assert rd.active_interval_s > 0, "active_interval_s must be positive"
-    assert rd.idle_interval_s > rd.active_interval_s, (
-        "idle_interval_s must be greater than active_interval_s")
-    # flatten runs cleanly over the real rep set
-    flat = rd.flatten(resources)
-    assert flat

tests/test_end_to_end_v2.py

@@ -2,7 +2,7 @@ import pytest
 from tests.conftest import _load_device
 from samsung_appliance.registry.by_type import for_device, _type_key
 from samsung_appliance.registry.discovery import discover
-from samsung_appliance.registry.adapter import build_runtime_descriptor
+from samsung_appliance.registry.adapter import flatten, is_active
 
 
 @pytest.mark.parametrize('name,expected_type_key', [
@@ -22,11 +22,7 @@ def test_full_pipeline_v2(name, expected_type_key):
     bound = discover(resources, reg.capabilities, reg.pattern_capabilities)
     assert bound
 
-    rd = build_runtime_descriptor(bound, topic_prefix=f'samsung_{name}',
-                                   ha_prefix='homeassistant', device_name=name.title(),
-                                   model='M', name=name, default_port=49154)
-    assert rd.discovery_payloads
-    flat = rd.flatten(resources)
-    assert flat
-    assert rd.active_interval_s > 0
-    assert rd.idle_interval_s > rd.active_interval_s
+    state = flatten(bound, resources)
+    assert state
+    # is_active returns bool; just verify it runs without error
+    is_active(bound, resources)

tests/test_golden_regression.py

@@ -6,10 +6,10 @@ import pytest
 GOLDEN = Path(__file__).parent / 'fixtures' / 'golden'
 
 
-def _new_state_and_uids(name, resources):
+def _new_state_keys(name, resources):
     from samsung_appliance.registry.by_type import for_device
     from samsung_appliance.registry.discovery import discover
-    from samsung_appliance.registry.adapter import build_runtime_descriptor
+    from samsung_appliance.registry.adapter import flatten
     otn = resources.get('/otninformation/vs/0', {})
     one_ui = otn.get('swVersionInfo', {}).get('oneUiVersion', '')
     reg = for_device(one_ui) if one_ui else None
@@ -19,12 +19,8 @@ def _new_state_and_uids(name, resources):
     else:
         caps, pats = reg.capabilities, reg.pattern_capabilities
     bound = discover(resources, caps, pats)
-    rd = build_runtime_descriptor(
-        bound, topic_prefix=f'samsung_{name}', ha_prefix='homeassistant',
-        device_name=name.title(), model='M', name=name, default_port=49154)
-    state = rd.flatten(resources)
-    uids = [json.loads(p)['unique_id'] for _, p in rd.discovery_payloads]
-    return sorted(state.keys()), sorted(uids)
+    state = flatten(bound, resources)
+    return sorted(state.keys())
 
 
 @pytest.mark.parametrize('name,ip', [
@@ -35,17 +31,12 @@ def test_registry_reproduces_golden_state_keys(name, ip, request):
     from tests.conftest import _load_resources
     resources = _load_resources(ip)
     golden = json.loads((GOLDEN / f'{name}.json').read_text())
-    state_keys, uids = _new_state_and_uids(name, resources)
+    state_keys = _new_state_keys(name, resources)
     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))}"
     )
-    assert set(uids) == set(golden['discovery_unique_ids']), (
-        f"unique_ids mismatch:\n"
-        f"  extra:   {sorted(set(uids) - set(golden['discovery_unique_ids']))}\n"
-        f"  missing: {sorted(set(golden['discovery_unique_ids']) - set(uids))}"
-    )
 
 
 def test_resources_from_batch_preferred_over_flat():

tests/test_oven_capabilities.py

@@ -1,6 +1,5 @@
-"""Unit tests for oven-family capabilities and the cycle-active adapter wiring."""
+"""Unit tests for oven-family capabilities."""
 from samsung_appliance.registry.capabilities import oven
-from samsung_appliance.registry.adapter import build_runtime_descriptor
 from samsung_appliance.registry.discovery import discover
 
 
@@ -151,36 +150,3 @@ def test_cook_time_rejects_out_of_range():
     assert desc.write_fn(-1, {}) is None
     assert desc.write_fn(1440, {}) is None
 
-
-# ---------------------------------------------------------------------------
-# Adapter cycle_active_field wiring
-# ---------------------------------------------------------------------------
-
-def test_adapter_sets_cycle_active_field_for_oven():
-    """build_runtime_descriptor must set cycle_active_field when oven keys present."""
-    reg = {c.href: [c] for c in (oven.OVEN_OPERATIONAL_STATE, oven.OVEN_SETPOINT)}
-    resources = {
-        '/operational/state/vs/0': {'x.com.samsung.da.state': 'Run'},
-        '/temperatures/vs/0': {'x.com.samsung.da.items': [
-            {'x.com.samsung.da.current': '180', 'x.com.samsung.da.desired': '200'}
-        ]},
-    }
-    bound = discover(resources, reg)
-    rd = build_runtime_descriptor(
-        bound, topic_prefix='t', ha_prefix='homeassistant',
-        device_name='Oven', model='M', name='oven', default_port=49154)
-    assert rd.cycle_active_field == 'cycle_active'
-    flat = rd.flatten(resources)
-    assert flat['cycle_active'] is True     # 'Run' -> active
-
-
-def test_adapter_no_cycle_active_without_oven_keys():
-    """Non-oven appliances must not get cycle_active_field."""
-    from samsung_appliance.registry.capabilities import common
-    reg = {c.href: [c] for c in (common.KIDS_LOCK,)}
-    resources = {'/kidslock/vs/0': {'x.com.samsung.da.kidsLock': 'Ready'}}
-    bound = discover(resources, reg)
-    rd = build_runtime_descriptor(
-        bound, topic_prefix='t', ha_prefix='homeassistant',
-        device_name='Dishwasher', model='M', name='dishwasher', default_port=49154)
-    assert rd.cycle_active_field is None

tests/test_state_cache.py

@@ -1,15 +0,0 @@
-from samsung_appliance.state_cache import StateCache
-
-
-def test_state_cache_constructs_without_descriptor():
-    c = StateCache()
-    assert c.links == {}
-    assert c.descriptor_state == {}
-
-
-def test_on_observation_hook_fires():
-    c = StateCache()
-    seen = []
-    c.set_on_observation(lambda st, href, rep: seen.append((href, rep)))
-    c.apply_rep('/x/vs/0', {'a': 1}, source='seed')
-    assert seen == [('/x/vs/0', {'a': 1})]