From 1eb4418ce39d9d368d4f195702215a7854790633 Mon Sep 17 00:00:00 2001 From: Ivan Malison Date: Tue, 12 May 2026 18:03:10 -0700 Subject: [PATCH] Normalize legacy optic source alias --- pykefcontrol/kef_connector.py | 17 +++++++--- tests/test_kef_connector.py | 58 +++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 4 deletions(-) create mode 100644 tests/test_kef_connector.py diff --git a/pykefcontrol/kef_connector.py b/pykefcontrol/kef_connector.py index a860488..56bf8fa 100644 --- a/pykefcontrol/kef_connector.py +++ b/pykefcontrol/kef_connector.py @@ -8,6 +8,11 @@ _POST_MODELS = {"LS50WII", "LSXIILT", "LSXII", "LS60"} _MODEL_ALIASES = {"LS50W2": "LS50WII", "LSX2LT": "LSXIILT", "LSX2": "LSXII"} +_SOURCE_ALIASES = {"optic": "optical"} + + +def _normalize_source(source): + return _SOURCE_ALIASES.get(source, source) class KefConnector: @@ -390,7 +395,7 @@ def status(self, status): def source(self): """ Speaker source : standby (not powered on), - wifi, bluetooth, tv, optic, coaxial or analog + wifi, bluetooth, tv, optical, coaxial or analog """ payload = { "path": "settings:/kef/play/physicalSource", @@ -408,8 +413,10 @@ def source(self): def source(self, source): """ Set spaker source, if speaker in standby, it powers on the speaker. - Possible sources : wifi, bluetooth, tv, optic, coaxial or analog + Possible sources : wifi, bluetooth, tv, optical, coaxial or analog. + The legacy value "optic" is accepted as an alias for "optical". """ + source = _normalize_source(source) payload = { "path": "settings:/kef/play/physicalSource", "roles": "value", @@ -656,7 +663,9 @@ async def get_wifi_information(self): async def set_source(self, source): """Set spaker source, if speaker in standby, it powers on the speaker. - Possible sources : wifi, bluetooth, tv, optic, coaxial or analog""" + Possible sources : wifi, bluetooth, tv, optical, coaxial or analog. + The legacy value "optic" is accepted as an alias for "optical".""" + source = _normalize_source(source) payload = { "path": "settings:/kef/play/physicalSource", "roles": "value", @@ -930,7 +939,7 @@ async def song_status(self): @property async def source(self): - """Speaker soe : standby (not powered on), wifi, bluetooth, tv, optic, + """Speaker soe : standby (not powered on), wifi, bluetooth, tv, optical, coaxial or analog""" payload = { "path": "settings:/kef/play/physicalSource", diff --git a/tests/test_kef_connector.py b/tests/test_kef_connector.py new file mode 100644 index 0000000..e10e880 --- /dev/null +++ b/tests/test_kef_connector.py @@ -0,0 +1,58 @@ +import asyncio +import unittest +from unittest.mock import AsyncMock, patch + +from pykefcontrol.kef_connector import KefAsyncConnector, KefConnector + + +class SourceNormalizationTest(unittest.TestCase): + def test_sync_source_setter_normalizes_optic_alias(self): + speaker = object.__new__(KefConnector) + speaker.host = "192.0.2.1" + speaker._speaker_model = "LS50WII" + + with patch("pykefcontrol.kef_connector.requests.post") as post: + post.return_value.__enter__.return_value.json.return_value = {} + + speaker.source = "optic" + + payload = post.call_args.kwargs["json"] + self.assertEqual( + payload["value"], + {"type": "kefPhysicalSource", "kefPhysicalSource": "optical"}, + ) + + def test_sync_source_setter_keeps_optical_source(self): + speaker = object.__new__(KefConnector) + speaker.host = "192.0.2.1" + speaker._speaker_model = "LS50WII" + + with patch("pykefcontrol.kef_connector.requests.post") as post: + post.return_value.__enter__.return_value.json.return_value = {} + + speaker.source = "optical" + + payload = post.call_args.kwargs["json"] + self.assertEqual( + payload["value"], + {"type": "kefPhysicalSource", "kefPhysicalSource": "optical"}, + ) + + def test_async_set_source_normalizes_optic_alias(self): + async def run_test(): + speaker = object.__new__(KefAsyncConnector) + speaker._set_data = AsyncMock() + + await speaker.set_source("optic") + + payload = speaker._set_data.call_args.args[0] + self.assertEqual( + payload["value"], + {"type": "kefPhysicalSource", "kefPhysicalSource": "optical"}, + ) + + asyncio.run(run_test()) + + +if __name__ == "__main__": + unittest.main()