Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 13 additions & 4 deletions pykefcontrol/kef_connector.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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",
Expand All @@ -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",
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down
58 changes: 58 additions & 0 deletions tests/test_kef_connector.py
Original file line number Diff line number Diff line change
@@ -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()