From d5558f51c1d455216871a3ffa29a40ef2b4f693b Mon Sep 17 00:00:00 2001 From: danielpetrovic Date: Mon, 16 Mar 2026 13:40:28 +0100 Subject: [PATCH 1/6] Add speaker settings: volume management, hardware controls, and LED controls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds 48 new methods (24 sync + 24 async) across three areas: Volume management: - get/set_default_volume(input_source) — per-source startup volume - get_all_default_volumes() — all source defaults at once - get/set_volume_settings(max_volume, step, limit) - get/set_standby_volume_behavior(use_global) - get/set_startup_volume_enabled(enabled) Hardware settings: - get/set_standby_mode(mode) — standby timeout - get/set_startup_tone(enabled) — power-on chime - get/set_auto_switch_hdmi(enabled) - get/set_cable_mode(mode) - get/set_master_channel(channel) - get/set_wake_source(source) - get/set_subwoofer_wake_on_startup(enabled) - get/set_kw1_wake_on_startup(enabled) - get/set_usb_charging(enabled) - get/set_fixed_volume_mode(volume) - set_request(path, roles, value) — generic write counterpart to get_request() LED controls: - get/set_front_led(enabled) - get/set_standby_led(enabled) - get/set_top_panel_enabled(enabled) - get/set_top_panel_led(enabled) - get/set_top_panel_standby_led(enabled) All methods available in both KefConnector (sync) and KefAsyncConnector (async). Tested on LSX II, LSX II LT, and XIO hardware. Co-Authored-By: Claude Sonnet 4.6 --- .gitignore | 4 +- pykefcontrol/kef_connector.py | 2479 ++++++++++++++++++++++++++++----- 2 files changed, 2130 insertions(+), 353 deletions(-) diff --git a/.gitignore b/.gitignore index c8db9b0..712c5b3 100644 --- a/.gitignore +++ b/.gitignore @@ -129,4 +129,6 @@ dmypy.json .pyre/ # VScode specific -.vscode/ \ No newline at end of file +.vscode/ +# APK decompilation working directories +apk/ diff --git a/pykefcontrol/kef_connector.py b/pykefcontrol/kef_connector.py index a860488..06e72f3 100644 --- a/pykefcontrol/kef_connector.py +++ b/pykefcontrol/kef_connector.py @@ -87,6 +87,324 @@ def set_volume(self, volume): """ self.volume = volume + def get_default_volume(self, input_source): + """Get default volume for a specific input source. + + Args: + input_source (str): Input source name (wifi, bluetooth, optic, coaxial, usb, analog, tv) + + Returns: + int: Volume level (0-100) for the specified input + + Example: + volume = speaker.get_default_volume('wifi') # Returns 50 + """ + # Map input source to API path + source_map = { + 'wifi': 'Wifi', + 'bluetooth': 'Bluetooth', + 'optic': 'Optical', + 'optical': 'Optical', + 'coaxial': 'Coaxial', + 'usb': 'USB', + 'analog': 'Analogue', + 'analogue': 'Analogue', + 'tv': 'TV', + 'hdmi': 'TV' + } + + if input_source.lower() not in source_map: + raise ValueError(f"Invalid input source: {input_source}. Valid sources: {', '.join(source_map.keys())}") + + api_source = source_map[input_source.lower()] + payload = { + "path": f"settings:/kef/host/defaultVolume{api_source}", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + return json_output[0]["i32_"] + + def set_default_volume(self, input_source, volume): + """Set default volume for a specific input source. + + Args: + input_source (str): Input source name (wifi, bluetooth, optic, coaxial, usb, analog, tv) + volume (int): Volume level (0-100) + + Example: + speaker.set_default_volume('wifi', 50) + speaker.set_default_volume('bluetooth', 40) + """ + if not 0 <= volume <= 100: + raise ValueError(f"Volume must be between 0 and 100, got {volume}") + + # Map input source to API path + source_map = { + 'global': 'Global', + 'wifi': 'Wifi', + 'bluetooth': 'Bluetooth', + 'optic': 'Optical', + 'optical': 'Optical', + 'coaxial': 'Coaxial', + 'usb': 'USB', + 'analog': 'Analogue', + 'analogue': 'Analogue', + 'tv': 'TV', + 'hdmi': 'TV' + } + + if input_source.lower() not in source_map: + raise ValueError(f"Invalid input source: {input_source}. Valid sources: {', '.join(source_map.keys())}") + + api_source = source_map[input_source.lower()] + payload = { + "path": f"settings:/kef/host/defaultVolume{api_source}", + "roles": "value", + "value": f'{{"type":"i32_","i32_":{volume}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_all_default_volumes(self): + """Get default volumes for all input sources on this speaker model. + + Returns: + dict: Dictionary of input sources and their default volumes + + Example: + volumes = speaker.get_all_default_volumes() + # Returns: {'global': 50, 'wifi': 45, 'bluetooth': 40, 'optical': 50, ...} + """ + # Define all possible inputs + all_inputs = ['global', 'wifi', 'bluetooth', 'optical', 'coaxial', 'usb', 'analogue', 'tv'] + + # Map to API names + source_map = { + 'global': 'Global', + 'wifi': 'Wifi', + 'bluetooth': 'Bluetooth', + 'optical': 'Optical', + 'coaxial': 'Coaxial', + 'usb': 'USB', + 'analogue': 'Analogue', + 'tv': 'TV' + } + + volumes = {} + for input_source in all_inputs: + try: + api_source = source_map[input_source] + payload = { + "path": f"settings:/kef/host/defaultVolume{api_source}", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + if response.status_code == 200: + json_output = response.json() + volumes[input_source] = json_output[0]["i32_"] + except: + # Skip inputs that don't exist on this model + pass + + return volumes + + def get_volume_settings(self): + """Get volume behavior settings. + + Returns: + dict: Volume settings including max_volume, step, limit, display mode + + Example: + settings = speaker.get_volume_settings() + # Returns: {'max_volume': 100, 'step': 1, 'limit': 100, 'display': 'linear'} + """ + settings = {} + + # Get maximum volume + try: + payload = {"path": "settings:/kef/host/maximumVolume", "roles": "value"} + with requests.get("http://" + self.host + "/api/getData", params=payload) as response: + if response.status_code == 200: + settings['max_volume'] = response.json()[0]["i32_"] + except: + pass + + # Get volume step (uses i16_ not i32_) + try: + payload = {"path": "settings:/kef/host/volumeStep", "roles": "value"} + with requests.get("http://" + self.host + "/api/getData", params=payload) as response: + if response.status_code == 200: + settings['step'] = response.json()[0]["i16_"] + except: + pass + + # Get volume limit (is bool, not int) + try: + payload = {"path": "settings:/kef/host/volumeLimit", "roles": "value"} + with requests.get("http://" + self.host + "/api/getData", params=payload) as response: + if response.status_code == 200: + settings['limit_enabled'] = response.json()[0]["bool_"] + except: + pass + + # Get volume display (XIO only) + try: + payload = {"path": "settings:/kef/host/volumeDisplay", "roles": "value"} + with requests.get("http://" + self.host + "/api/getData", params=payload) as response: + if response.status_code == 200: + settings['display'] = response.json()[0]["string_"] + except: + pass + + return settings + + def set_volume_settings(self, max_volume=None, step=None, limit=None): + """Set volume behavior settings. + + Args: + max_volume (int, optional): Maximum volume (0-100) + step (int, optional): Volume increment step + limit (int, optional): Volume limiter (0-100) + + Example: + speaker.set_volume_settings(max_volume=80, step=2) + speaker.set_volume_settings(limit=75) + """ + if max_volume is not None: + if not 0 <= max_volume <= 100: + raise ValueError(f"max_volume must be between 0 and 100, got {max_volume}") + payload = { + "path": "settings:/kef/host/maximumVolume", + "roles": "value", + "value": f'{{"type":"i32_","i32_":{max_volume}}}', + } + with requests.get("http://" + self.host + "/api/setData", params=payload) as response: + pass + + if step is not None: + if not 1 <= step <= 10: + raise ValueError(f"step must be between 1 and 10, got {step}") + payload = { + "path": "settings:/kef/host/volumeStep", + "roles": "value", + "value": f'{{"type":"i16_","i16_":{step}}}', + } + with requests.get("http://" + self.host + "/api/setData", params=payload) as response: + pass + + if limit is not None: + payload = { + "path": "settings:/kef/host/volumeLimit", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(limit).lower()}}}', + } + with requests.get("http://" + self.host + "/api/setData", params=payload) as response: + pass + + def get_standby_volume_behavior(self): + """Get standby volume behavior setting. + + Returns: + bool: True if using global volume mode, False if using per-input mode + + Example: + is_global = speaker.get_standby_volume_behavior() + """ + payload = { + "path": "settings:/kef/host/advancedStandbyDefaultVol", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + # advancedStandbyDefaultVol: false = global, true = per-input + return not json_output[0]["bool_"] + + def set_standby_volume_behavior(self, use_global): + """Set standby volume behavior. + + Args: + use_global (bool): True for global volume mode, False for per-input mode + + Example: + speaker.set_standby_volume_behavior(True) # Use global volume + speaker.set_standby_volume_behavior(False) # Use per-input volumes + """ + # advancedStandbyDefaultVol: false = global, true = per-input + payload = { + "path": "settings:/kef/host/advancedStandbyDefaultVol", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(not use_global).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_startup_volume_enabled(self): + """Get whether startup volume feature is enabled. + + When enabled, the speaker uses configured startup volumes when waking from standby. + When disabled, the speaker resumes at the last volume level. + + Returns: + bool: True if startup volume is enabled, False if disabled + + Example: + is_enabled = speaker.get_startup_volume_enabled() + """ + payload = { + "path": "settings:/kef/host/standbyDefaultVol", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + return json_output[0]["bool_"] + + def set_startup_volume_enabled(self, enabled): + """Enable or disable the startup volume feature. + + When enabled, the speaker uses configured startup volumes when waking from standby. + When disabled, the speaker resumes at the last volume level. + + Args: + enabled (bool): True to enable startup volume, False to disable + + Example: + speaker.set_startup_volume_enabled(True) # Enable startup volume + speaker.set_startup_volume_enabled(False) # Disable (resume at last volume) + """ + payload = { + "path": "settings:/kef/host/standbyDefaultVol", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + # Network Diagnostics Methods (Phase 4) def _get_player_data(self): """ Is the speaker currently playing @@ -513,481 +831,1938 @@ def firmware_version(self): return speaker_firmware_version -class KefAsyncConnector: - def __init__(self, host, session=None, model=None): - self.host = host - self._session = session - self._speaker_model = _MODEL_ALIASES.get(model, model) - self.previous_volume = ( - 15 # Hardcoded previous volume, in case unmute is used before mute - ) - self.last_polled = None - self.polling_queue = None - self._previous_poll_song_status = False - - async def close_session(self): - """close session""" - if self._session is not None: - await self._session.close() - self._session = None + def get_auto_switch_hdmi(self): + """Get auto-switch to HDMI setting. - async def resurect_session(self): - if self._session is None: - self._session = aiohttp.ClientSession() + Returns: + bool: True if auto-switch enabled, False otherwise - async def power_on(self): - """power on speaker""" - await self.set_status("powerOn") + Example: + enabled = speaker.get_auto_switch_hdmi() + """ + payload = { + "path": "settings:/kef/host/autoSwitchToHDMI", + "roles": "value", + } - async def shutdown(self): - """Shutdown speaker""" - await self.set_source("standby") + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() - async def mute(self): - """mute speaker""" - self.previous_volume = await self.volume - await self.set_volume(0) + return json_output[0].get("bool_", False) - async def unmute(self): - """unmute speaker""" - await self.set_volume(self.previous_volume) + def set_auto_switch_hdmi(self, enabled): + """Set auto-switch to HDMI when signal detected. - async def toggle_play_pause(self): - """Toogle play/pause""" - await self._track_control("pause") + Args: + enabled (bool): True to enable auto-switch, False to disable - async def next_track(self): - """Next track""" - await self._track_control("next") + Example: + speaker.set_auto_switch_hdmi(True) + """ + payload = { + "path": "settings:/kef/host/autoSwitchToHDMI", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } - async def previous_track(self): - """Previous track""" - await self._track_control("previous") + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() - async def _set_data(self, payload): - if self._speaker_model is None: - self._speaker_model = await self.get_speaker_model() - await self.resurect_session() - if self._speaker_model in _POST_MODELS: - async with self._session.post( - "http://" + self.host + "/api/setData", json=payload - ) as response: - return await response.json() - else: - payload = dict(payload) - payload["value"] = json.dumps(payload["value"], separators=(",", ":")) - async with self._session.get( - "http://" + self.host + "/api/setData", params=payload - ) as response: - return await response.json() + def get_standby_mode(self): + """Get auto-standby mode setting. - async def _track_control(self, command): - """toogle play/pause""" - payload = { - "path": "player:player/control", - "roles": "activate", - "value": {"control": command}, - } - await self._set_data(payload) + Returns: + str: Standby mode ('standby_20mins', 'standby_30mins', 'standby_60mins', 'standby_none') - async def _get_player_data(self): - """get data about currently playing media""" + Example: + mode = speaker.get_standby_mode() # Returns 'standby_20mins' (ECO mode) + """ payload = { - "path": "player:player/data", + "path": "settings:/kef/host/standbyMode", "roles": "value", } - await self.resurect_session() - async with self._session.get( + + with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: - json_output = await response.json() + json_output = response.json() - return json_output[0] + return json_output[0].get("string_", "standby_20mins") - async def get_request(self, path, roles="value"): - """Generic method to get data from any API path. + def set_standby_mode(self, mode): + """Set auto-standby mode. Args: - path: API path to query (e.g., "kef:eqProfile", "network:info") - roles: API roles parameter (default: "value") + mode (str): Standby mode - 'standby_20mins' (ECO), 'standby_30mins', + 'standby_60mins', or 'standby_none' (Never) - Returns: - JSON response from API + Example: + speaker.set_standby_mode('standby_20mins') # ECO mode (20 minutes) + speaker.set_standby_mode('standby_none') # Never auto-standby """ + valid_modes = ['standby_20mins', 'standby_30mins', 'standby_60mins', 'standby_none'] + if mode not in valid_modes: + raise ValueError(f"Invalid mode: {mode}. Valid modes: {', '.join(valid_modes)}") + payload = { - "path": path, - "roles": roles, + "path": "settings:/kef/host/standbyMode", + "roles": "value", + "value": f'{{"type":"string_","string_":"{mode}"}}', } - await self.resurect_session() - async with self._session.get( - "http://" + self.host + "/api/getData", params=payload + + with requests.get( + "http://" + self.host + "/api/setData", params=payload ) as response: - json_output = await response.json() + json_output = response.json() - return json_output + def get_startup_tone(self): + """Get startup tone setting. - async def get_wifi_information(self): - """Get WiFi information from speaker. + Returns: + bool: True if startup beep enabled, False otherwise - Returns dict with WiFi signal strength, SSID, frequency, and BSSID. - Returns empty dict if WiFi info is not available. + Example: + enabled = speaker.get_startup_tone() """ - try: - # Get network info from speaker - network_data = await self.get_request("network:info", roles="value") - - wifi_dict = {} - network_info = ( - network_data[0].get("networkInfo", {}) if network_data else {} - ) - - if network_info: - wireless = network_info.get("wireless", {}) - if wireless: - wifi_dict["signalLevel"] = wireless.get("signalLevel") - wifi_dict["ssid"] = wireless.get("ssid") - wifi_dict["frequency"] = wireless.get("frequency") - wifi_dict["bssid"] = wireless.get("bssid") - - return wifi_dict - except Exception: - # Silently return empty dict if WiFi info not available - return {} - - 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""" payload = { - "path": "settings:/kef/play/physicalSource", + "path": "settings:/kef/host/startupTone", "roles": "value", - "value": {"type": "kefPhysicalSource", "kefPhysicalSource": source}, } - await self._set_data(payload) + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + return json_output[0].get("bool_", False) + + def set_startup_tone(self, enabled): + """Set startup tone (power-on beep). + + Args: + enabled (bool): True to enable startup beep, False to disable + + Example: + speaker.set_startup_tone(False) # Disable startup beep + """ + payload = { + "path": "settings:/kef/host/startupTone", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_subwoofer_wake_on_startup(self): + """Get wake subwoofer on startup setting. + + When enabled, the speaker will wake the subwoofer when it powers on. + This works with wired subwoofers. + + Returns: + bool: True if wake subwoofer on startup is enabled + """ + payload = { + "path": "settings:/kef/host/subwooferForceOn", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + return json_output[0].get("bool_", False) + + def set_subwoofer_wake_on_startup(self, enabled): + """Set wake subwoofer on startup. + + When enabled, the speaker will wake the subwoofer when it powers on. + This works with wired subwoofers. + + Args: + enabled (bool): True to enable wake subwoofer on startup + + Example: + speaker.set_subwoofer_wake_on_startup(True) + """ + payload = { + "path": "settings:/kef/host/subwooferForceOn", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_kw1_wake_on_startup(self): + """Get KW1 wake on startup setting. + + When enabled, the speaker will wake a wireless subwoofer connected + via KW1 adapter when it powers on. This is specifically for + KC62/KF92 subwoofers with KW1 wireless adapter. + + Returns: + bool: True if KW1 wake on startup is enabled + """ + payload = { + "path": "settings:/kef/host/subwooferForceOnKW1", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + return json_output[0].get("bool_", False) + + def set_kw1_wake_on_startup(self, enabled): + """Set KW1 wake on startup. + + When enabled, the speaker will wake a wireless subwoofer connected + via KW1 adapter when it powers on. This is specifically for + KC62/KF92 subwoofers with KW1 wireless adapter. + + Args: + enabled (bool): True to enable KW1 wake on startup + + Example: + speaker.set_kw1_wake_on_startup(True) + """ + payload = { + "path": "settings:/kef/host/subwooferForceOnKW1", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_wake_source(self): + """Get wake-up source setting. + + Returns: + str: Wake source ('wakeup_default', 'tv', 'wifi', 'bluetooth', 'optical') + + Example: + source = speaker.get_wake_source() # Returns 'wakeup_default' + """ + payload = { + "path": "settings:/kef/host/wakeUpSource", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + return json_output[0].get("kefWakeUpSource", "wakeup_default") + + def set_wake_source(self, source): + """Set wake-up source. + + Args: + source (str): Wake source - 'wakeup_default', 'tv', 'wifi', 'bluetooth', 'optical' + + Example: + speaker.set_wake_source('tv') # Wake on TV/HDMI signal + """ + valid_sources = ['wakeup_default', 'tv', 'wifi', 'bluetooth', 'optical'] + if source not in valid_sources: + raise ValueError(f"Invalid source: {source}. Valid sources: {', '.join(valid_sources)}") + + payload = { + "path": "settings:/kef/host/wakeUpSource", + "roles": "value", + "value": f'{{"type":"kefWakeUpSource","kefWakeUpSource":"{source}"}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_usb_charging(self): + """Get USB charging setting. + + Returns: + bool: True if USB charging enabled, False otherwise + + Example: + enabled = speaker.get_usb_charging() + """ + payload = { + "path": "settings:/kef/host/usbCharging", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + return json_output[0].get("bool_", False) + + def set_usb_charging(self, enabled): + """Set USB port charging. + + Args: + enabled (bool): True to enable USB charging, False to disable + + Example: + speaker.set_usb_charging(True) # Enable USB charging + """ + payload = { + "path": "settings:/kef/host/usbCharging", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_cable_mode(self): + """Get cable mode (wired/wireless inter-speaker connection). + + Returns: + str: Cable mode ('wired' or 'wireless') + + Example: + mode = speaker.get_cable_mode() # Returns 'wired' + """ + payload = { + "path": "settings:/kef/host/cableMode", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + return json_output[0].get("string_", "wired") + + def set_cable_mode(self, mode): + """Set cable mode for inter-speaker connection. + + Args: + mode (str): Cable mode - 'wired' or 'wireless' + + Example: + speaker.set_cable_mode('wireless') # Use wireless connection + """ + valid_modes = ['wired', 'wireless'] + if mode not in valid_modes: + raise ValueError(f"Invalid mode: {mode}. Valid modes: {', '.join(valid_modes)}") + + payload = { + "path": "settings:/kef/host/cableMode", + "roles": "value", + "value": f'{{"type":"string_","string_":"{mode}"}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_master_channel(self): + """Get master channel (left/right speaker designation). + + Returns: + str: Master channel ('left' or 'right') + + Example: + channel = speaker.get_master_channel() # Returns 'left' + """ + payload = { + "path": "settings:/kef/host/masterChannelMode", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + return json_output[0].get("kefMasterChannelMode", "right") + + def set_master_channel(self, channel): + """Set master channel designation. + + Args: + channel (str): Master channel - 'left' or 'right' + + Example: + speaker.set_master_channel('right') # Set as right speaker + """ + valid_channels = ['left', 'right'] + if channel not in valid_channels: + raise ValueError(f"Invalid channel: {channel}. Valid channels: {', '.join(valid_channels)}") + + payload = { + "path": "settings:/kef/host/masterChannelMode", + "roles": "value", + "value": f'{{"type":"kefMasterChannelMode","kefMasterChannelMode":"{channel}"}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + + def get_front_led(self): + """Get front panel LED setting. + + Note: This API setting exists but has no visible effect on any + currently tested KEF speakers (LSX II, LSX II LT, XIO). The setting + may be reserved for future models or have no hardware implementation. + + Returns: + bool: True if front LED is enabled, False if disabled + + Example: + enabled = speaker.get_front_led() + """ + payload = { + "path": "settings:/kef/host/disableFrontLED", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + # Note: API uses "disable" so we invert the boolean + return not json_output[0].get("bool_", False) + + def set_front_led(self, enabled): + """Set front panel LED. + + Note: This API setting exists but has no visible effect on any + currently tested KEF speakers (LSX II, LSX II LT, XIO). The setting + may be reserved for future models or have no hardware implementation. + + Args: + enabled (bool): True to enable LED, False to disable + + Example: + speaker.set_front_led(False) # Disable front LED + """ + # Note: API uses "disable" so we invert the boolean + disabled = not enabled + payload = { + "path": "settings:/kef/host/disableFrontLED", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(disabled).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_standby_led(self): + """Get standby LED setting. + + Returns: + bool: True if standby LED is enabled, False if disabled + + Example: + enabled = speaker.get_standby_led() + """ + payload = { + "path": "settings:/kef/host/disableFrontStandbyLED", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + # Note: API uses "disable" so we invert the boolean + return not json_output[0].get("bool_", False) + + def set_standby_led(self, enabled): + """Set standby LED. + + Args: + enabled (bool): True to enable LED, False to disable + + Example: + speaker.set_standby_led(True) # Enable standby LED + """ + # Note: API uses "disable" so we invert the boolean + disabled = not enabled + payload = { + "path": "settings:/kef/host/disableFrontStandbyLED", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(disabled).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_top_panel_enabled(self): + """Get top panel (touch controls) enabled setting. + + Returns: + bool: True if top panel is enabled, False if disabled + + Example: + enabled = speaker.get_top_panel_enabled() + """ + payload = { + "path": "settings:/kef/host/disableTopPanel", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + + # Note: API uses "disable" so we invert the boolean + return not json_output[0].get("bool_", False) + + def set_top_panel_enabled(self, enabled): + """Set top panel (touch controls) enabled. + + Args: + enabled (bool): True to enable top panel, False to disable + + Example: + speaker.set_top_panel_enabled(False) # Disable touch panel + """ + # Note: API uses "disable" so we invert the boolean + disabled = not enabled + payload = { + "path": "settings:/kef/host/disableTopPanel", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(disabled).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_top_panel_led(self): + """Get top panel LED setting (XIO only). + + Returns: + bool: True if enabled, False if disabled, None if not available (non-XIO speakers) + + Example: + enabled = speaker.get_top_panel_led() # XIO only + if enabled is not None: + print(f"Top panel LED: {'ON' if enabled else 'OFF'}") + """ + payload = { + "path": "settings:/kef/host/topPanelLED", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + # Check if response is an error (dict with 'error' key) or empty + if isinstance(json_output, dict) and 'error' in json_output: + return None + if json_output and len(json_output) > 0: + return json_output[0].get("bool_", False) + return None + + def set_top_panel_led(self, enabled): + """Set top panel LED (XIO only). + + Args: + enabled (bool): True to enable LED, False to disable + + Example: + speaker.set_top_panel_led(True) # XIO only + """ + payload = { + "path": "settings:/kef/host/topPanelLED", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + def get_top_panel_standby_led(self): + """Get top panel standby LED setting (XIO only). + + Returns: + bool: True if enabled, False if disabled, None if not available (non-XIO speakers) + + Example: + enabled = speaker.get_top_panel_standby_led() # XIO only + if enabled is not None: + print(f"Top panel standby LED: {'ON' if enabled else 'OFF'}") + """ + payload = { + "path": "settings:/kef/host/topPanelStandbyLED", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + # Check if response is an error (dict with 'error' key) or empty + if isinstance(json_output, dict) and 'error' in json_output: + return None + if json_output and len(json_output) > 0: + return json_output[0].get("bool_", False) + return None + + def set_top_panel_standby_led(self, enabled): + """Set top panel standby LED (XIO only). + + Args: + enabled (bool): True to enable LED, False to disable + + Example: + speaker.set_top_panel_standby_led(False) # XIO only + """ + payload = { + "path": "settings:/kef/host/topPanelStandbyLED", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + # ===== Remote Control Methods ===== + + + def get_fixed_volume_mode(self): + """Get fixed volume mode setting. + + Returns: + int or None: Fixed volume level (0-100), or None if disabled + + Example: + volume = speaker.get_fixed_volume_mode() + if volume is not None: + print(f"Fixed volume: {volume}") + else: + print("Fixed volume mode disabled") + """ + payload = { + "path": "settings:/kef/host/remote/userFixedVolume", + "roles": "value", + } + + with requests.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = response.json() + value = json_output[0].get("i32_", -1) + return None if value < 0 else value + + def set_fixed_volume_mode(self, volume): + """Set fixed volume mode (locks volume at specific level). + + Args: + volume (int or None): Volume level to lock (0-100), or None to disable + + Example: + speaker.set_fixed_volume_mode(50) # Lock volume at 50% + speaker.set_fixed_volume_mode(None) # Disable fixed volume mode + """ + if volume is None: + volume = -1 # -1 disables fixed volume mode + elif not isinstance(volume, int) or volume < 0 or volume > 100: + raise ValueError("Volume must be between 0-100 or None to disable") + + payload = { + "path": "settings:/kef/host/remote/userFixedVolume", + "roles": "value", + "value": f'{{"type":"i32_","i32_":{volume}}}', + } + + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + + # Generic write method + def set_request(self, path, roles="value", value=None): + """Generic method to set data via any API path. + + Args: + path: API path to set (e.g., "firmwareupdate:downloadNewUpdate") + roles: API roles parameter (default: "value", use "activate" for actions) + value: Optional value to set (can be JSON string or dict) + + Returns: + JSON response from API + """ + payload = { + "path": path, + "roles": roles, + } + if value is not None: + payload["value"] = value + with requests.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = response.json() + + return json_output + + +class KefAsyncConnector: + def __init__(self, host, session=None, model=None): + self.host = host + self._session = session + self._speaker_model = _MODEL_ALIASES.get(model, model) + self.previous_volume = ( + 15 # Hardcoded previous volume, in case unmute is used before mute + ) + self.last_polled = None + self.polling_queue = None + self._previous_poll_song_status = False + + async def close_session(self): + """close session""" + if self._session is not None: + await self._session.close() + self._session = None + + async def resurect_session(self): + if self._session is None: + self._session = aiohttp.ClientSession() + + async def power_on(self): + """power on speaker""" + await self.set_status("powerOn") + + async def shutdown(self): + """Shutdown speaker""" + await self.set_source("standby") + + async def mute(self): + """mute speaker""" + self.previous_volume = await self.volume + await self.set_volume(0) + + async def unmute(self): + """unmute speaker""" + await self.set_volume(self.previous_volume) + + async def toggle_play_pause(self): + """Toogle play/pause""" + await self._track_control("pause") + + async def next_track(self): + """Next track""" + await self._track_control("next") + + async def previous_track(self): + """Previous track""" + await self._track_control("previous") + + async def _set_data(self, payload): + if self._speaker_model is None: + self._speaker_model = await self.get_speaker_model() + await self.resurect_session() + if self._speaker_model in _POST_MODELS: + async with self._session.post( + "http://" + self.host + "/api/setData", json=payload + ) as response: + return await response.json() + else: + payload = dict(payload) + payload["value"] = json.dumps(payload["value"], separators=(",", ":")) + async with self._session.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + return await response.json() + + async def _track_control(self, command): + """toogle play/pause""" + payload = { + "path": "player:player/control", + "roles": "activate", + "value": {"control": command}, + } + await self._set_data(payload) + + async def _get_player_data(self): + """get data about currently playing media""" + payload = { + "path": "player:player/data", + "roles": "value", + } + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + return json_output[0] + + async def get_request(self, path, roles="value"): + """Generic method to get data from any API path. + + Args: + path: API path to query (e.g., "kef:eqProfile", "network:info") + roles: API roles parameter (default: "value") + + Returns: + JSON response from API + """ + payload = { + "path": path, + "roles": roles, + } + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + return json_output + + async def set_request(self, path, roles="value", value=None): + """Generic method to set data via any API path. + + Args: + path: API path to set (e.g., "firmwareupdate:install") + roles: API roles parameter (default: "value") + value: Optional value to send (JSON string) + + Returns: + JSON response from API + """ + payload = { + "path": path, + "roles": roles, + } + if value is not None: + payload["value"] = value + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = await response.json() + + return json_output + + async def get_wifi_information(self): + """Get WiFi information from speaker. + + Returns dict with WiFi signal strength, SSID, frequency, and BSSID. + Returns empty dict if WiFi info is not available. + """ + try: + # Get network info from speaker + network_data = await self.get_request("network:info", roles="value") + + wifi_dict = {} + network_info = ( + network_data[0].get("networkInfo", {}) if network_data else {} + ) + + if network_info: + wireless = network_info.get("wireless", {}) + if wireless: + wifi_dict["signalLevel"] = wireless.get("signalLevel") + wifi_dict["ssid"] = wireless.get("ssid") + wifi_dict["frequency"] = wireless.get("frequency") + wifi_dict["bssid"] = wireless.get("bssid") + + return wifi_dict + except Exception: + # Silently return empty dict if WiFi info not available + return {} + + 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""" + payload = { + "path": "settings:/kef/play/physicalSource", + "roles": "value", + "value": {"type": "kefPhysicalSource", "kefPhysicalSource": source}, + } + await self._set_data(payload) async def set_volume(self, volume): """Set speaker volume (between 0 and 100)""" payload = { "path": "player:volume", "roles": "value", - "value": {"type": "i32_", "i32_": volume}, + "value": {"type": "i32_", "i32_": volume}, + } + await self._set_data(payload) + + async def set_status(self, status): + payload = { + "path": "settings:/kef/play/physicalSource", + "roles": "value", + "value": {"type": "kefPhysicalSource", "kefPhysicalSource": status}, + } + await self._set_data(payload) + + async def get_song_information(self, song_data=None): + """Get song title, album and artist""" + if song_data == None: + song_data = await self._get_player_data() + info_dict = dict() + info_dict["title"] = song_data.get("trackRoles", {}).get("title") + + metadata = ( + song_data.get("trackRoles", {}) + .get("mediaData", {}) + .get("metaData", {}) + ) + + info_dict["artist"] = metadata.get("artist") + info_dict["album"] = metadata.get("album") + # Use albumArtist if available, otherwise fallback to artist + album_artist = metadata.get("albumArtist") + info_dict["album_artist"] = album_artist if album_artist else metadata.get("artist") + info_dict["cover_url"] = song_data.get("trackRoles", {}).get("icon", None) + info_dict["service_id"] = metadata.get("serviceID") + + return info_dict + + async def get_audio_codec_information(self, player_data=None): + """ + Get audio codec information from player data. + Returns dict with codec, sample rate, and channel information. + """ + try: + if player_data is None: + player_data = await self._get_player_data() + + codec_dict = {} + active_resource = ( + player_data.get("trackRoles", {}) + .get("mediaData", {}) + .get("activeResource", {}) + ) + + if active_resource: + codec_dict["codec"] = active_resource.get("codec") + codec_dict["sampleFrequency"] = active_resource.get("sampleFrequency") + codec_dict["streamSampleRate"] = active_resource.get("streamSampleRate") + codec_dict["streamChannels"] = active_resource.get("streamChannels") + codec_dict["nrAudioChannels"] = active_resource.get("nrAudioChannels") + + # Get streaming service ID from metadata + metadata = ( + player_data.get("trackRoles", {}) + .get("mediaData", {}) + .get("metaData", {}) + ) + if metadata: + codec_dict["serviceID"] = metadata.get("serviceID") + + return codec_dict + except Exception: + # Silently return empty dict if codec info not available + return {} + + async def get_polling_queue(self, song_status=False, poll_song_status=False): + """Get the polling queue uuid, and subscribe to all relevant topics""" + payload = { + "subscribe": [ + {"path": "settings:/mediaPlayer/playMode", "type": "itemWithValue"}, + {"path": "playlists:pq/getitems", "type": "rows"}, + {"path": "notifications:/display/queue", "type": "rows"}, + {"path": "settings:/kef/host/maximumVolume", "type": "itemWithValue"}, + {"path": "player:volume", "type": "itemWithValue"}, + {"path": "kef:fwupgrade/info", "type": "itemWithValue"}, + {"path": "settings:/kef/host/volumeStep", "type": "itemWithValue"}, + {"path": "settings:/kef/host/volumeLimit", "type": "itemWithValue"}, + {"path": "settings:/mediaPlayer/mute", "type": "itemWithValue"}, + {"path": "settings:/kef/host/speakerStatus", "type": "itemWithValue"}, + {"path": "settings:/kef/play/physicalSource", "type": "itemWithValue"}, + {"path": "player:player/data", "type": "itemWithValue"}, + {"path": "kef:speedTest/status", "type": "itemWithValue"}, + {"path": "network:info", "type": "itemWithValue"}, + {"path": "kef:eqProfile", "type": "itemWithValue"}, + {"path": "settings:/kef/host/modelName", "type": "itemWithValue"}, + {"path": "settings:/version", "type": "itemWithValue"}, + {"path": "settings:/deviceName", "type": "itemWithValue"}, + ], + "unsubscribe": [], + } + + if song_status: + payload["subscribe"].append( + {"path": "player:player/data/playTime", "type": "itemWithValue"} + ) + + await self.resurect_session() + async with self._session.post( + "http://" + self.host + "/api/event/modifyQueue", json=payload + ) as response: + json_output = await response.json() + + # Update polling_queue property with queue uuid + self.polling_queue = json_output[1:-1] + + # Update last polled time + self.last_polled = time.time() + + return self.polling_queue + + async def parse_events(self, events): + """Parse events""" + parsed_events = dict() + + for event in events: + if event == "settings:/kef/play/physicalSource": + parsed_events["source"] = events[event].get("kefPhysicalSource") + elif event == "player:player/data/playTime": + parsed_events["song_status"] = events[event].get("i64_") + elif event == "player:volume": + parsed_events["volume"] = events[event].get("i32_") + elif event == "player:player/data": + parsed_events["song_info"] = await self.get_song_information( + events[event] + ) + parsed_events["song_length"] = ( + events[event].get("status", {}).get("duration") + ) + parsed_events["status"] = events[event].get("state") + elif event == "settings:/kef/host/speakerStatus": + parsed_events["speaker_status"] = events[event].get("kefSpeakerStatus") + elif event == "settings:/deviceName": + parsed_events["device_name"] = events[event].get("string_") + elif event == "settings:/mediaPlayer/mute": + parsed_events["mute"] = events[event].get("bool_") + else: + if parsed_events.get("other") == None: + parsed_events["other"] = {} + parsed_events["other"].update({event: events[event]}) + + return parsed_events + + async def poll_speaker(self, timeout=10, song_status=False, poll_song_status=False): + """poll speaker for info""" + + if song_status: + warnings.warn( + "The 'song_status' parameter is deprecated and will be removed in version 0.8.0. " + "Please use 'poll_song_status' instead.", + DeprecationWarning, + stacklevel=2, + ) + + # check if it is necessary to get a new queue + if ( + (self.polling_queue == None) + or ((time.time() - self.last_polled) > 50) + or (song_status != self._previous_polling_song_status) + ): + await self.get_polling_queue(poll_song_status=poll_song_status) + + payload = {"queueId": "{{{}}}".format(self.polling_queue), "timeout": timeout} + + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/event/pollQueue", + params=payload, + timeout=10 + 0.5, # add 0.5 seconds to timeout to allow for processing + ) as response: + json_output = await response.json() + + # Process all events + + events = dict() + # fill events lists + for j in json_output: + if events.get(j["path"], False): + events[j["path"]].append(j) + else: + events[j["path"]] = [j] + # prune events lists + for k in events: + events[k] = events[k][-1].get("itemValue", "updated") + + return await self.parse_events(events) + + @property + async def mac_address(self): + """Get the mac address of the Speaker""" + payload = {"path": "settings:/system/primaryMacAddress", "roles": "value"} + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + return json_output[0]["string_"] + + @property + async def speaker_name(self): + """Get the friendly name of the Speaker""" + payload = {"path": "settings:/deviceName", "roles": "value"} + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + return json_output[0]["string_"] + + @property + async def status(self): + """Status of the speaker : standby or poweredOn""" + payload = {"path": "settings:/kef/host/speakerStatus", "roles": "value"} + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + return json_output[0]["kefSpeakerStatus"] + + @property + async def is_playing(self): + """Is the speaker currently playing""" + json_output = await self._get_player_data() + return json_output["state"] == "playing" + + @property + async def song_length(self): + """Song length in ms""" + if await self.is_playing: + json_output = await self._get_player_data() + return json_output["status"]["duration"] + else: + return None + + @property + async def song_status(self): + """Progression of song""" + payload = { + "path": "player:player/data/playTime", + "roles": "value", + } + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + return json_output[0]["i64_"] + + @property + async def source(self): + """Speaker soe : standby (not powered on), wifi, bluetooth, tv, optic, + coaxial or analog""" + payload = { + "path": "settings:/kef/play/physicalSource", + "roles": "value", + } + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + return json_output[0]["kefPhysicalSource"] + + @property + async def volume(self): + """Speaker volume (1 to 100, 0 = muted)""" + payload = { + "path": "player:volume", + "roles": "value", } - await self._set_data(payload) + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + return json_output[0]["i32_"] + + async def _get_speaker_firmware_version(self): + """ + Get speaker firmware "release text" + """ - async def set_status(self, status): payload = { - "path": "settings:/kef/play/physicalSource", + "path": "settings:/releasetext", "roles": "value", - "value": {"type": "kefPhysicalSource", "kefPhysicalSource": status}, } - await self._set_data(payload) + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() - async def get_song_information(self, song_data=None): - """Get song title, album and artist""" - if song_data == None: - song_data = await self._get_player_data() - info_dict = dict() - info_dict["title"] = song_data.get("trackRoles", {}).get("title") + return json_output[0]["string_"] - metadata = ( - song_data.get("trackRoles", {}) - .get("mediaData", {}) - .get("metaData", {}) - ) + async def get_speaker_model(self): + """ + Speaker model + """ + raw_data = await self._get_speaker_firmware_version() + speaker_model = raw_data.split("_")[0] + return speaker_model - info_dict["artist"] = metadata.get("artist") - info_dict["album"] = metadata.get("album") - # Use albumArtist if available, otherwise fallback to artist - album_artist = metadata.get("albumArtist") - info_dict["album_artist"] = album_artist if album_artist else metadata.get("artist") - info_dict["cover_url"] = song_data.get("trackRoles", {}).get("icon", None) - info_dict["service_id"] = metadata.get("serviceID") + async def get_firmware_version(self): + """ + Speaker firmware version + """ + raw_data = await self._get_speaker_firmware_version() + speaker_firmware_version = raw_data.split("_")[1] + return speaker_firmware_version - return info_dict + async def get_default_volume(self, input_source): + """Get default volume for a specific input source. - async def get_audio_codec_information(self, player_data=None): + Args: + input_source (str): Input source name (wifi, bluetooth, optic, coaxial, usb, analog, tv) + + Returns: + int: Volume level (0-100) for the specified input + + Example: + volume = await speaker.get_default_volume('wifi') # Returns 50 """ - Get audio codec information from player data. - Returns dict with codec, sample rate, and channel information. + # Map input source to API path + source_map = { + 'wifi': 'Wifi', + 'bluetooth': 'Bluetooth', + 'optic': 'Optical', + 'optical': 'Optical', + 'coaxial': 'Coaxial', + 'usb': 'USB', + 'analog': 'Analogue', + 'analogue': 'Analogue', + 'tv': 'TV', + 'hdmi': 'TV' + } + + if input_source.lower() not in source_map: + raise ValueError(f"Invalid input source: {input_source}. Valid sources: {', '.join(source_map.keys())}") + + api_source = source_map[input_source.lower()] + payload = { + "path": f"settings:/kef/host/defaultVolume{api_source}", + "roles": "value", + } + + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + return json_output[0]["i32_"] + + async def set_default_volume(self, input_source, volume): + """Set default volume for a specific input source. + + Args: + input_source (str): Input source name (global, wifi, bluetooth, optic, coaxial, usb, analog, tv) + volume (int): Volume level (0-100) + + Example: + await speaker.set_default_volume('global', 30) # Set global startup volume + await speaker.set_default_volume('wifi', 50) + await speaker.set_default_volume('bluetooth', 40) + """ + if not 0 <= volume <= 100: + raise ValueError(f"Volume must be between 0 and 100, got {volume}") + + # Map input source to API path + source_map = { + 'global': 'Global', + 'wifi': 'Wifi', + 'bluetooth': 'Bluetooth', + 'optic': 'Optical', + 'optical': 'Optical', + 'coaxial': 'Coaxial', + 'usb': 'USB', + 'analog': 'Analogue', + 'analogue': 'Analogue', + 'tv': 'TV', + 'hdmi': 'TV' + } + + if input_source.lower() not in source_map: + raise ValueError(f"Invalid input source: {input_source}. Valid sources: {', '.join(source_map.keys())}") + + api_source = source_map[input_source.lower()] + payload = { + "path": f"settings:/kef/host/defaultVolume{api_source}", + "roles": "value", + "value": f'{{"type":"i32_","i32_":{volume}}}', + } + + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = await response.json() + + async def get_all_default_volumes(self): + """Get default volumes for all input sources on this speaker model. + + Returns: + dict: Dictionary of input sources and their default volumes + + Example: + volumes = await speaker.get_all_default_volumes() + # Returns: {'global': 50, 'wifi': 45, 'bluetooth': 40, 'optical': 50, ...} + """ + # Define all possible inputs + all_inputs = ['global', 'wifi', 'bluetooth', 'optical', 'coaxial', 'usb', 'analogue', 'tv'] + + # Map to API names + source_map = { + 'global': 'Global', + 'wifi': 'Wifi', + 'bluetooth': 'Bluetooth', + 'optical': 'Optical', + 'coaxial': 'Coaxial', + 'usb': 'USB', + 'analogue': 'Analogue', + 'tv': 'TV' + } + + volumes = {} + await self.resurect_session() + + for input_source in all_inputs: + try: + api_source = source_map[input_source] + payload = { + "path": f"settings:/kef/host/defaultVolume{api_source}", + "roles": "value", + } + + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + if response.status == 200: + json_output = await response.json() + volumes[input_source] = json_output[0]["i32_"] + except: + # Skip inputs that don't exist on this model + pass + + return volumes + + async def get_volume_settings(self): + """Get volume behavior settings. + + Returns: + dict: Volume settings including max_volume, step, limit, display mode + + Example: + settings = await speaker.get_volume_settings() + # Returns: {'max_volume': 100, 'step': 1, 'limit': 100, 'display': 'linear'} """ + settings = {} + await self.resurect_session() + + # Get maximum volume try: - if player_data is None: - player_data = await self._get_player_data() + payload = {"path": "settings:/kef/host/maximumVolume", "roles": "value"} + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + if response.status == 200: + json_output = await response.json() + settings['max_volume'] = json_output[0]["i32_"] + except: + pass + + # Get volume step (uses i16_ not i32_) + try: + payload = {"path": "settings:/kef/host/volumeStep", "roles": "value"} + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + if response.status == 200: + json_output = await response.json() + settings['step'] = json_output[0]["i16_"] + except: + pass + + # Get volume limit (is bool, not int) + try: + payload = {"path": "settings:/kef/host/volumeLimit", "roles": "value"} + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + if response.status == 200: + json_output = await response.json() + settings['limit_enabled'] = json_output[0]["bool_"] + except: + pass + + # Get volume display (XIO only) + try: + payload = {"path": "settings:/kef/host/volumeDisplay", "roles": "value"} + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + if response.status == 200: + json_output = await response.json() + settings['display'] = json_output[0]["string_"] + except: + pass - codec_dict = {} - active_resource = ( - player_data.get("trackRoles", {}) - .get("mediaData", {}) - .get("activeResource", {}) - ) + return settings - if active_resource: - codec_dict["codec"] = active_resource.get("codec") - codec_dict["sampleFrequency"] = active_resource.get("sampleFrequency") - codec_dict["streamSampleRate"] = active_resource.get("streamSampleRate") - codec_dict["streamChannels"] = active_resource.get("streamChannels") - codec_dict["nrAudioChannels"] = active_resource.get("nrAudioChannels") + async def set_volume_settings(self, max_volume=None, step=None, limit=None): + """Set volume behavior settings. - # Get streaming service ID from metadata - metadata = ( - player_data.get("trackRoles", {}) - .get("mediaData", {}) - .get("metaData", {}) - ) - if metadata: - codec_dict["serviceID"] = metadata.get("serviceID") + Args: + max_volume (int, optional): Maximum volume (0-100) + step (int, optional): Volume increment step (1-10) + limit (bool, optional): Enable volume limiter - return codec_dict - except Exception: - # Silently return empty dict if codec info not available - return {} + Example: + await speaker.set_volume_settings(max_volume=80, step=2) + await speaker.set_volume_settings(limit=True) + """ + await self.resurect_session() - async def get_polling_queue(self, song_status=False, poll_song_status=False): - """Get the polling queue uuid, and subscribe to all relevant topics""" + if max_volume is not None: + if not 0 <= max_volume <= 100: + raise ValueError(f"max_volume must be between 0 and 100, got {max_volume}") + payload = { + "path": "settings:/kef/host/maximumVolume", + "roles": "value", + "value": f'{{"type":"i32_","i32_":{max_volume}}}', + } + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + pass + + if step is not None: + if not 1 <= step <= 10: + raise ValueError(f"step must be between 1 and 10, got {step}") + payload = { + "path": "settings:/kef/host/volumeStep", + "roles": "value", + "value": f'{{"type":"i16_","i16_":{step}}}', + } + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + pass + + if limit is not None: + payload = { + "path": "settings:/kef/host/volumeLimit", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(limit).lower()}}}', + } + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + pass + + async def get_standby_volume_behavior(self): + """Get standby volume behavior setting. + + Returns: + bool: True if using global volume mode (All sources), False if using per-input mode (Individual sources) + + Example: + is_global = await speaker.get_standby_volume_behavior() + """ payload = { - "subscribe": [ - {"path": "settings:/mediaPlayer/playMode", "type": "itemWithValue"}, - {"path": "playlists:pq/getitems", "type": "rows"}, - {"path": "notifications:/display/queue", "type": "rows"}, - {"path": "settings:/kef/host/maximumVolume", "type": "itemWithValue"}, - {"path": "player:volume", "type": "itemWithValue"}, - {"path": "kef:fwupgrade/info", "type": "itemWithValue"}, - {"path": "settings:/kef/host/volumeStep", "type": "itemWithValue"}, - {"path": "settings:/kef/host/volumeLimit", "type": "itemWithValue"}, - {"path": "settings:/mediaPlayer/mute", "type": "itemWithValue"}, - {"path": "settings:/kef/host/speakerStatus", "type": "itemWithValue"}, - {"path": "settings:/kef/play/physicalSource", "type": "itemWithValue"}, - {"path": "player:player/data", "type": "itemWithValue"}, - {"path": "kef:speedTest/status", "type": "itemWithValue"}, - {"path": "network:info", "type": "itemWithValue"}, - {"path": "kef:eqProfile", "type": "itemWithValue"}, - {"path": "settings:/kef/host/modelName", "type": "itemWithValue"}, - {"path": "settings:/version", "type": "itemWithValue"}, - {"path": "settings:/deviceName", "type": "itemWithValue"}, - ], - "unsubscribe": [], + "path": "settings:/kef/host/advancedStandbyDefaultVol", + "roles": "value", + } + + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + # advancedStandbyDefaultVol: false = global (All sources), true = per-input (Individual sources) + return not json_output[0]["bool_"] + + async def set_standby_volume_behavior(self, use_global): + """Set standby volume behavior. + + Args: + use_global (bool): True for global volume mode (All sources), False for per-input mode (Individual sources) + + Example: + await speaker.set_standby_volume_behavior(True) # All sources + await speaker.set_standby_volume_behavior(False) # Individual sources + """ + # advancedStandbyDefaultVol: false = global (All sources), true = per-input (Individual sources) + payload = { + "path": "settings:/kef/host/advancedStandbyDefaultVol", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(not use_global).lower()}}}', + } + + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = await response.json() + + async def get_startup_volume_enabled(self): + """Get whether reset volume feature is enabled. + + When enabled, the speaker uses configured reset volumes when waking from standby. + When disabled, the speaker resumes at the last volume level. + + Returns: + bool: True if reset volume is enabled, False if disabled + + Example: + is_enabled = await speaker.get_startup_volume_enabled() + """ + payload = { + "path": "settings:/kef/host/standbyDefaultVol", + "roles": "value", + } + + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/getData", params=payload + ) as response: + json_output = await response.json() + + return json_output[0]["bool_"] + + async def set_startup_volume_enabled(self, enabled): + """Enable or disable the reset volume feature. + + When enabled, the speaker uses configured reset volumes when waking from standby. + When disabled, the speaker resumes at the last volume level. + + Args: + enabled (bool): True to enable reset volume, False to disable + + Example: + await speaker.set_startup_volume_enabled(True) # Enable reset volume + await speaker.set_startup_volume_enabled(False) # Disable (resume at last volume) + """ + payload = { + "path": "settings:/kef/host/standbyDefaultVol", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + + await self.resurect_session() + async with self._session.get( + "http://" + self.host + "/api/setData", params=payload + ) as response: + json_output = await response.json() + + # Async Network Diagnostics Methods (Phase 4) + + async def get_auto_switch_hdmi(self): + """Get auto-switch to HDMI setting.""" + payload = {"path": "settings:/kef/host/autoSwitchToHDMI", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + return json_output[0].get("bool_", False) + + async def set_auto_switch_hdmi(self, enabled): + """Set auto-switch to HDMI when signal detected.""" + payload = { + "path": "settings:/kef/host/autoSwitchToHDMI", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', } + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + json_output = await response.json() - if song_status: - payload["subscribe"].append( - {"path": "player:player/data/playTime", "type": "itemWithValue"} - ) + async def get_standby_mode(self): + """Get auto-standby mode setting.""" + payload = {"path": "settings:/kef/host/standbyMode", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + return json_output[0].get("string_", "standby_20mins") + async def set_standby_mode(self, mode): + """Set auto-standby mode.""" + valid_modes = ['standby_20mins', 'standby_30mins', 'standby_60mins', 'standby_none'] + if mode not in valid_modes: + raise ValueError(f"Invalid mode: {mode}. Valid modes: {', '.join(valid_modes)}") + payload = { + "path": "settings:/kef/host/standbyMode", + "roles": "value", + "value": f'{{"type":"string_","string_":"{mode}"}}', + } await self.resurect_session() - async with self._session.post( - "http://" + self.host + "/api/event/modifyQueue", json=payload - ) as response: + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: json_output = await response.json() - # Update polling_queue property with queue uuid - self.polling_queue = json_output[1:-1] + async def get_startup_tone(self): + """Get startup tone setting.""" + payload = {"path": "settings:/kef/host/startupTone", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + return json_output[0].get("bool_", False) - # Update last polled time - self.last_polled = time.time() + async def set_startup_tone(self, enabled): + """Set startup tone (power-on beep).""" + payload = { + "path": "settings:/kef/host/startupTone", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + json_output = await response.json() - return self.polling_queue + async def get_subwoofer_wake_on_startup(self): + """Get wake subwoofer on startup setting. - async def parse_events(self, events): - """Parse events""" - parsed_events = dict() + When enabled, the speaker will wake the subwoofer when it powers on. + This works with wired subwoofers. - for event in events: - if event == "settings:/kef/play/physicalSource": - parsed_events["source"] = events[event].get("kefPhysicalSource") - elif event == "player:player/data/playTime": - parsed_events["song_status"] = events[event].get("i64_") - elif event == "player:volume": - parsed_events["volume"] = events[event].get("i32_") - elif event == "player:player/data": - parsed_events["song_info"] = await self.get_song_information( - events[event] - ) - parsed_events["song_length"] = ( - events[event].get("status", {}).get("duration") - ) - parsed_events["status"] = events[event].get("state") - elif event == "settings:/kef/host/speakerStatus": - parsed_events["speaker_status"] = events[event].get("kefSpeakerStatus") - elif event == "settings:/deviceName": - parsed_events["device_name"] = events[event].get("string_") - elif event == "settings:/mediaPlayer/mute": - parsed_events["mute"] = events[event].get("bool_") - else: - if parsed_events.get("other") == None: - parsed_events["other"] = {} - parsed_events["other"].update({event: events[event]}) + Returns: + bool: True if wake subwoofer on startup is enabled + """ + payload = {"path": "settings:/kef/host/subwooferForceOn", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + return json_output[0].get("bool_", False) - return parsed_events + async def set_subwoofer_wake_on_startup(self, enabled): + """Set wake subwoofer on startup. - async def poll_speaker(self, timeout=10, song_status=False, poll_song_status=False): - """poll speaker for info""" + When enabled, the speaker will wake the subwoofer when it powers on. + This works with wired subwoofers. - if song_status: - warnings.warn( - "The 'song_status' parameter is deprecated and will be removed in version 0.8.0. " - "Please use 'poll_song_status' instead.", - DeprecationWarning, - stacklevel=2, - ) + Args: + enabled (bool): True to enable wake subwoofer on startup + """ + payload = { + "path": "settings:/kef/host/subwooferForceOn", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + json_output = await response.json() - # check if it is necessary to get a new queue - if ( - (self.polling_queue == None) - or ((time.time() - self.last_polled) > 50) - or (song_status != self._previous_polling_song_status) - ): - await self.get_polling_queue(poll_song_status=poll_song_status) + async def get_kw1_wake_on_startup(self): + """Get KW1 wake on startup setting. - payload = {"queueId": "{{{}}}".format(self.polling_queue), "timeout": timeout} + When enabled, the speaker will wake a wireless subwoofer connected + via KW1 adapter when it powers on. This is specifically for + KC62/KF92 subwoofers with KW1 wireless adapter. + Returns: + bool: True if KW1 wake on startup is enabled + """ + payload = {"path": "settings:/kef/host/subwooferForceOnKW1", "roles": "value"} await self.resurect_session() - async with self._session.get( - "http://" + self.host + "/api/event/pollQueue", - params=payload, - timeout=10 + 0.5, # add 0.5 seconds to timeout to allow for processing - ) as response: + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: json_output = await response.json() + return json_output[0].get("bool_", False) - # Process all events + async def set_kw1_wake_on_startup(self, enabled): + """Set KW1 wake on startup. - events = dict() - # fill events lists - for j in json_output: - if events.get(j["path"], False): - events[j["path"]].append(j) - else: - events[j["path"]] = [j] - # prune events lists - for k in events: - events[k] = events[k][-1].get("itemValue", "updated") + When enabled, the speaker will wake a wireless subwoofer connected + via KW1 adapter when it powers on. This is specifically for + KC62/KF92 subwoofers with KW1 wireless adapter. - return await self.parse_events(events) + Args: + enabled (bool): True to enable KW1 wake on startup + """ + payload = { + "path": "settings:/kef/host/subwooferForceOnKW1", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + json_output = await response.json() - @property - async def mac_address(self): - """Get the mac address of the Speaker""" - payload = {"path": "settings:/system/primaryMacAddress", "roles": "value"} + async def get_wake_source(self): + """Get wake-up source setting.""" + payload = {"path": "settings:/kef/host/wakeUpSource", "roles": "value"} await self.resurect_session() - async with self._session.get( - "http://" + self.host + "/api/getData", params=payload - ) as response: + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: json_output = await response.json() + return json_output[0].get("kefWakeUpSource", "wakeup_default") - return json_output[0]["string_"] + async def set_wake_source(self, source): + """Set wake-up source.""" + valid_sources = ['wakeup_default', 'tv', 'wifi', 'bluetooth', 'optical'] + if source not in valid_sources: + raise ValueError(f"Invalid source: {source}. Valid sources: {', '.join(valid_sources)}") + payload = { + "path": "settings:/kef/host/wakeUpSource", + "roles": "value", + "value": f'{{"type":"kefWakeUpSource","kefWakeUpSource":"{source}"}}', + } + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + json_output = await response.json() - @property - async def speaker_name(self): - """Get the friendly name of the Speaker""" - payload = {"path": "settings:/deviceName", "roles": "value"} + async def get_usb_charging(self): + """Get USB charging setting.""" + payload = {"path": "settings:/kef/host/usbCharging", "roles": "value"} await self.resurect_session() - async with self._session.get( - "http://" + self.host + "/api/getData", params=payload - ) as response: + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: json_output = await response.json() + return json_output[0].get("bool_", False) - return json_output[0]["string_"] + async def set_usb_charging(self, enabled): + """Set USB port charging.""" + payload = { + "path": "settings:/kef/host/usbCharging", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + json_output = await response.json() - @property - async def status(self): - """Status of the speaker : standby or poweredOn""" - payload = {"path": "settings:/kef/host/speakerStatus", "roles": "value"} + async def get_cable_mode(self): + """Get cable mode (wired/wireless inter-speaker connection).""" + payload = {"path": "settings:/kef/host/cableMode", "roles": "value"} await self.resurect_session() - async with self._session.get( - "http://" + self.host + "/api/getData", params=payload - ) as response: + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: json_output = await response.json() + return json_output[0].get("string_", "wired") - return json_output[0]["kefSpeakerStatus"] + async def set_cable_mode(self, mode): + """Set cable mode for inter-speaker connection.""" + valid_modes = ['wired', 'wireless'] + if mode not in valid_modes: + raise ValueError(f"Invalid mode: {mode}. Valid modes: {', '.join(valid_modes)}") + payload = { + "path": "settings:/kef/host/cableMode", + "roles": "value", + "value": f'{{"type":"string_","string_":"{mode}"}}', + } + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + json_output = await response.json() - @property - async def is_playing(self): - """Is the speaker currently playing""" - json_output = await self._get_player_data() - return json_output["state"] == "playing" + async def get_master_channel(self): + """Get master channel (left/right speaker designation).""" + payload = {"path": "settings:/kef/host/masterChannelMode", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + return json_output[0].get("kefMasterChannelMode", "right") - @property - async def song_length(self): - """Song length in ms""" - if await self.is_playing: - json_output = await self._get_player_data() - return json_output["status"]["duration"] - else: - return None + async def set_master_channel(self, channel): + """Set master channel designation.""" + valid_channels = ['left', 'right'] + if channel not in valid_channels: + raise ValueError(f"Invalid channel: {channel}. Valid channels: {', '.join(valid_channels)}") + payload = { + "path": "settings:/kef/host/masterChannelMode", + "roles": "value", + "value": f'{{"type":"kefMasterChannelMode","kefMasterChannelMode":"{channel}"}}', + } + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + json_output = await response.json() - @property - async def song_status(self): - """Progression of song""" + + async def get_front_led(self): + """Get front panel LED setting. + + Note: This API setting exists but has no visible effect on any + currently tested KEF speakers (LSX II, LSX II LT, XIO). + """ + payload = {"path": "settings:/kef/host/disableFrontLED", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + return not json_output[0].get("bool_", False) + + async def set_front_led(self, enabled): + """Set front panel LED. + + Note: This API setting exists but has no visible effect on any + currently tested KEF speakers (LSX II, LSX II LT, XIO). + """ + disabled = not enabled payload = { - "path": "player:player/data/playTime", + "path": "settings:/kef/host/disableFrontLED", "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(disabled).lower()}}}', } await self.resurect_session() - async with self._session.get( - "http://" + self.host + "/api/getData", params=payload - ) as response: + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: json_output = await response.json() - return json_output[0]["i64_"] + async def get_standby_led(self): + """Get standby LED setting.""" + payload = {"path": "settings:/kef/host/disableFrontStandbyLED", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + return not json_output[0].get("bool_", False) - @property - async def source(self): - """Speaker soe : standby (not powered on), wifi, bluetooth, tv, optic, - coaxial or analog""" + async def set_standby_led(self, enabled): + """Set standby LED.""" + disabled = not enabled payload = { - "path": "settings:/kef/play/physicalSource", + "path": "settings:/kef/host/disableFrontStandbyLED", "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(disabled).lower()}}}', } await self.resurect_session() - async with self._session.get( - "http://" + self.host + "/api/getData", params=payload - ) as response: + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: json_output = await response.json() - return json_output[0]["kefPhysicalSource"] + async def get_top_panel_enabled(self): + """Get top panel (touch controls) enabled setting.""" + payload = {"path": "settings:/kef/host/disableTopPanel", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + return not json_output[0].get("bool_", False) - @property - async def volume(self): - """Speaker volume (1 to 100, 0 = muted)""" + async def set_top_panel_enabled(self, enabled): + """Set top panel (touch controls) enabled.""" + disabled = not enabled payload = { - "path": "player:volume", + "path": "settings:/kef/host/disableTopPanel", "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(disabled).lower()}}}', } await self.resurect_session() - async with self._session.get( - "http://" + self.host + "/api/getData", params=payload - ) as response: + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: json_output = await response.json() - return json_output[0]["i32_"] + async def get_top_panel_led(self): + """Get top panel LED setting (XIO only). - async def _get_speaker_firmware_version(self): - """ - Get speaker firmware "release text" + Returns: + bool: True if enabled, False if disabled, None if not available (non-XIO speakers) """ - + payload = {"path": "settings:/kef/host/topPanelLED", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + # Check if response is an error (dict with 'error' key) or empty + if isinstance(json_output, dict) and 'error' in json_output: + return None + if json_output and len(json_output) > 0: + return json_output[0].get("bool_", False) + return None + + async def set_top_panel_led(self, enabled): + """Set top panel LED (XIO only).""" payload = { - "path": "settings:/releasetext", + "path": "settings:/kef/host/topPanelLED", "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', } await self.resurect_session() - async with self._session.get( - "http://" + self.host + "/api/getData", params=payload - ) as response: + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: json_output = await response.json() - return json_output[0]["string_"] + async def get_top_panel_standby_led(self): + """Get top panel standby LED setting (XIO only). - async def get_speaker_model(self): - """ - Speaker model + Returns: + bool: True if enabled, False if disabled, None if not available (non-XIO speakers) """ - raw_data = await self._get_speaker_firmware_version() - speaker_model = raw_data.split("_")[0] - return speaker_model + payload = {"path": "settings:/kef/host/topPanelStandbyLED", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + # Check if response is an error (dict with 'error' key) or empty + if isinstance(json_output, dict) and 'error' in json_output: + return None + if json_output and len(json_output) > 0: + return json_output[0].get("bool_", False) + return None + + async def set_top_panel_standby_led(self, enabled): + """Set top panel standby LED (XIO only).""" + payload = { + "path": "settings:/kef/host/topPanelStandbyLED", + "roles": "value", + "value": f'{{"type":"bool_","bool_":{str(enabled).lower()}}}', + } + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + json_output = await response.json() - async def get_firmware_version(self): + # ===== Remote Control Methods (Async) ===== + + + async def get_fixed_volume_mode(self): + """Get fixed volume mode setting. + + Returns: + int or None: Fixed volume level (0-100), or None if disabled + + Example: + volume = await speaker.get_fixed_volume_mode() + if volume is not None: + print(f"Fixed volume: {volume}") + else: + print("Fixed volume mode disabled") """ - Speaker firmware version + payload = {"path": "settings:/kef/host/remote/userFixedVolume", "roles": "value"} + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + json_output = await response.json() + value = json_output[0].get("i32_", -1) + return None if value < 0 else value + + async def set_fixed_volume_mode(self, volume): + """Set fixed volume mode (locks volume at specific level). + + Args: + volume (int or None): Volume level to lock (0-100), or None to disable + + Example: + await speaker.set_fixed_volume_mode(50) # Lock volume at 50% + await speaker.set_fixed_volume_mode(None) # Disable fixed volume mode """ - raw_data = await self._get_speaker_firmware_version() - speaker_firmware_version = raw_data.split("_")[1] - return speaker_firmware_version + if volume is None: + volume = -1 # -1 disables fixed volume mode + elif not isinstance(volume, int) or volume < 0 or volume > 100: + raise ValueError("Volume must be between 0-100 or None to disable") + + payload = { + "path": "settings:/kef/host/remote/userFixedVolume", + "roles": "value", + "value": f'{{"type":"i32_","i32_":{volume}}}', + } + await self.resurect_session() + async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + json_output = await response.json() + From 1d004d8df7b53d20e37e3944563a2b75993fe3c5 Mon Sep 17 00:00:00 2001 From: danielpetrovic Date: Mon, 16 Mar 2026 13:53:51 +0100 Subject: [PATCH 2/6] Add README documentation for volume, hardware, and LED methods Document all 24 new methods added in this PR with usage examples: - Volume management: default volumes per source, volume settings, standby behavior - Hardware settings: standby mode, startup tone, HDMI, cable mode, wake source, LEDs - LED controls: front LED, standby LED, top panel enable/LED states Co-Authored-By: Claude Sonnet 4.6 --- README.md | 105 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 105 insertions(+) diff --git a/README.md b/README.md index b1d9130..b178109 100644 --- a/README.md +++ b/README.md @@ -204,6 +204,111 @@ All the possible keys of the dictionary are: `source`, `song_status`, `volume`, `song_info`, `song_length`, `status`, `speaker_status`, `device_name`, `mute` and `other`. `other` contains some of the speaker-specific information that might have changed, but are not properties of either `KefConnector` or `KefAsyncConnector`. +**Volume Management** +```python +# Get default startup volume for a specific input source +my_speaker.get_default_volume('wifi') +# (output example) >>> 50 + +# Set default startup volume for a specific input source +# Supported sources: 'wifi', 'bluetooth', 'optical', 'coaxial', 'usb', 'analogue', 'tv' +my_speaker.set_default_volume('wifi', 50) + +# Get default volumes for all input sources at once +my_speaker.get_all_default_volumes() +# (output example) >>> {'wifi': 50, 'bluetooth': 40, 'optical': 60, 'coaxial': 60, 'usb': 50, 'analogue': 60, 'tv': 50} + +# Get volume settings (max volume cap, step size, limit) +my_speaker.get_volume_settings() +# (output example) >>> {'max_volume': 100, 'step': 5, 'limit': 85} + +# Set volume settings +my_speaker.set_volume_settings(max_volume=100, step=5, limit=85) + +# Get/set whether standby resumes at global or per-source volume +my_speaker.get_standby_volume_behavior() +# (output example) >>> True # True = use global volume, False = use per-source volume +my_speaker.set_standby_volume_behavior(True) + +# Get/set startup volume override (when enabled, uses configured default volumes on wake) +my_speaker.get_startup_volume_enabled() +# (output example) >>> False +my_speaker.set_startup_volume_enabled(True) +``` + +**Hardware Settings** +```python +# Standby timeout +my_speaker.get_standby_mode() +# (output example) >>> 'standbyOff' # 'standbyOff', 'standby20m', 'standby60m' +my_speaker.set_standby_mode('standby20m') + +# Power-on chime +my_speaker.get_startup_tone() +# (output example) >>> True +my_speaker.set_startup_tone(False) + +# HDMI auto-source switching +my_speaker.get_auto_switch_hdmi() +# (output example) >>> True +my_speaker.set_auto_switch_hdmi(False) + +# Wired connection mode +my_speaker.get_cable_mode() +my_speaker.set_cable_mode('stereo') + +# Left/right master channel assignment +my_speaker.get_master_channel() +# (output example) >>> 'left' +my_speaker.set_master_channel('right') + +# Which input source wakes the speaker from standby +my_speaker.get_wake_source() +my_speaker.set_wake_source('wifi') + +# Subwoofer and KW1 wireless sub auto-wake on startup +my_speaker.get_subwoofer_wake_on_startup() +my_speaker.set_subwoofer_wake_on_startup(True) +my_speaker.get_kw1_wake_on_startup() +my_speaker.set_kw1_wake_on_startup(True) + +# USB port charging +my_speaker.get_usb_charging() +my_speaker.set_usb_charging(True) + +# Fixed volume output level (for use with AV receivers) +my_speaker.get_fixed_volume_mode() +# (output example) >>> None # None = disabled, integer = fixed level +my_speaker.set_fixed_volume_mode(50) + +# Generic write method (counterpart to get_request) +my_speaker.set_request('settings:/kef/host/speakerOn', 'activate', 'true') +``` + +**LED Controls** +```python +# Front logo LED +my_speaker.get_front_led() +# (output example) >>> True +my_speaker.set_front_led(False) + +# Standby indicator LED +my_speaker.get_standby_led() +my_speaker.set_standby_led(True) + +# Top touch panel (enable/disable) +my_speaker.get_top_panel_enabled() +my_speaker.set_top_panel_enabled(True) + +# Top panel LED in active state +my_speaker.get_top_panel_led() +my_speaker.set_top_panel_led(True) + +# Top panel LED in standby state +my_speaker.get_top_panel_standby_led() +my_speaker.set_top_panel_standby_led(False) +``` + #### Advanced features This function is used internally by pykefcontrol and returns a JSON output with a lot of information. You might want to use them to get extra information such as the artwork/album cover URL, which does not have a dedicated function _yet_ in pykefcontrol. From 4f52c06199a80d00e6689ceddea9df7b651397ed Mon Sep 17 00:00:00 2001 From: danielpetrovic Date: Mon, 16 Mar 2026 13:56:45 +0100 Subject: [PATCH 3/6] Update README with comprehensive documentation for new methods Replace simple stubs with full documentation sections matching the detail level of the rest of the library: - Volume Management: per-input defaults, volume settings, reset volume, all-sources vs individual-sources mode, async examples - System Behavior Settings: standby mode, wake source, HDMI auto-switch, startup tone, USB charging, cable mode, master channel, async examples - Do Not Disturb Settings: standby LED, startup tone, XIO control panel LEDs (front LED, top panel enable/standby), async examples Co-Authored-By: Claude Sonnet 4.6 --- README.md | 445 +++++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 355 insertions(+), 90 deletions(-) diff --git a/README.md b/README.md index b178109..c52d3ca 100644 --- a/README.md +++ b/README.md @@ -204,111 +204,376 @@ All the possible keys of the dictionary are: `source`, `song_status`, `volume`, `song_info`, `song_length`, `status`, `speaker_status`, `device_name`, `mute` and `other`. `other` contains some of the speaker-specific information that might have changed, but are not properties of either `KefConnector` or `KefAsyncConnector`. -**Volume Management** +### Volume Management + +Control per-input default volumes and volume behavior settings. Each physical input (WiFi, Bluetooth, Optical, etc.) can have its own default volume level, or you can use a global volume for all inputs. + +#### Per-Input Default Volumes + +Set different default volumes for each input source: + ```python -# Get default startup volume for a specific input source -my_speaker.get_default_volume('wifi') -# (output example) >>> 50 - -# Set default startup volume for a specific input source -# Supported sources: 'wifi', 'bluetooth', 'optical', 'coaxial', 'usb', 'analogue', 'tv' -my_speaker.set_default_volume('wifi', 50) - -# Get default volumes for all input sources at once -my_speaker.get_all_default_volumes() -# (output example) >>> {'wifi': 50, 'bluetooth': 40, 'optical': 60, 'coaxial': 60, 'usb': 50, 'analogue': 60, 'tv': 50} - -# Get volume settings (max volume cap, step size, limit) -my_speaker.get_volume_settings() -# (output example) >>> {'max_volume': 100, 'step': 5, 'limit': 85} - -# Set volume settings -my_speaker.set_volume_settings(max_volume=100, step=5, limit=85) - -# Get/set whether standby resumes at global or per-source volume -my_speaker.get_standby_volume_behavior() -# (output example) >>> True # True = use global volume, False = use per-source volume -my_speaker.set_standby_volume_behavior(True) - -# Get/set startup volume override (when enabled, uses configured default volumes on wake) -my_speaker.get_startup_volume_enabled() -# (output example) >>> False -my_speaker.set_startup_volume_enabled(True) +import pykefcontrol as pkf + +speaker = pkf.KefConnector('192.168.1.100') + +# Get default volume for a specific input +wifi_volume = speaker.get_default_volume('wifi') # Returns 0-100 +bluetooth_volume = speaker.get_default_volume('bluetooth') + +# Set default volume for specific inputs +speaker.set_default_volume('wifi', 50) # Set WiFi to 50% +speaker.set_default_volume('bluetooth', 40) # Set Bluetooth to 40% +speaker.set_default_volume('optical', 60) # Set Optical to 60% + +# Get all default volumes at once +all_volumes = speaker.get_all_default_volumes() +# Returns: {'global': 30, 'wifi': 50, 'bluetooth': 40, 'optical': 60, ...} + +# Print all volumes +for source, volume in sorted(all_volumes.items()): + print(f"{source:12s}: {volume}%") ``` -**Hardware Settings** +**Available input sources by model:** +- **LSX II**: wifi, bluetooth, optical, usb, analogue, tv (6 inputs) +- **LSX II LT**: wifi, bluetooth, optical, usb, tv (5 inputs) +- **LS50 Wireless II**: wifi, bluetooth, optical, coaxial, analogue, tv (6 inputs) +- **LS60 Wireless**: wifi, bluetooth, optical, coaxial, analogue, tv (6 inputs) +- **XIO Soundbar**: wifi, bluetooth, optical, tv (4 inputs) + +#### Volume Behavior Settings + +Configure global volume limits and behavior: + ```python -# Standby timeout -my_speaker.get_standby_mode() -# (output example) >>> 'standbyOff' # 'standbyOff', 'standby20m', 'standby60m' -my_speaker.set_standby_mode('standby20m') +# Get current volume settings +settings = speaker.get_volume_settings() +# Returns: {'max_volume': 100, 'step': 1, 'limit': 100, 'display': 'linear'} -# Power-on chime -my_speaker.get_startup_tone() -# (output example) >>> True -my_speaker.set_startup_tone(False) +# Set maximum volume limit (safety feature for children/hearing protection) +speaker.set_volume_settings(max_volume=80) # Limit to 80% -# HDMI auto-source switching -my_speaker.get_auto_switch_hdmi() -# (output example) >>> True -my_speaker.set_auto_switch_hdmi(False) - -# Wired connection mode -my_speaker.get_cable_mode() -my_speaker.set_cable_mode('stereo') - -# Left/right master channel assignment -my_speaker.get_master_channel() -# (output example) >>> 'left' -my_speaker.set_master_channel('right') - -# Which input source wakes the speaker from standby -my_speaker.get_wake_source() -my_speaker.set_wake_source('wifi') - -# Subwoofer and KW1 wireless sub auto-wake on startup -my_speaker.get_subwoofer_wake_on_startup() -my_speaker.set_subwoofer_wake_on_startup(True) -my_speaker.get_kw1_wake_on_startup() -my_speaker.set_kw1_wake_on_startup(True) - -# USB port charging -my_speaker.get_usb_charging() -my_speaker.set_usb_charging(True) - -# Fixed volume output level (for use with AV receivers) -my_speaker.get_fixed_volume_mode() -# (output example) >>> None # None = disabled, integer = fixed level -my_speaker.set_fixed_volume_mode(50) - -# Generic write method (counterpart to get_request) -my_speaker.set_request('settings:/kef/host/speakerOn', 'activate', 'true') +# Set volume step size (how much volume changes per button press) +speaker.set_volume_settings(step=2) # Change by 2% per step + +# Set volume limiter +speaker.set_volume_settings(limit=75) # Soft limit at 75% + +# Combine multiple settings +speaker.set_volume_settings(max_volume=85, step=2, limit=80) ``` -**LED Controls** +#### Reset Volume (Startup Volume) + +The "Reset Volume" feature (called "Startup Volume" in some contexts) controls what volume the speaker uses when waking from standby. This matches the KEF Connect app's "Reset volume" setting. + ```python -# Front logo LED -my_speaker.get_front_led() -# (output example) >>> True -my_speaker.set_front_led(False) +# Check if reset volume is enabled +is_enabled = speaker.get_startup_volume_enabled() +# Returns: True = enabled, False = disabled (resumes at last volume) -# Standby indicator LED -my_speaker.get_standby_led() -my_speaker.set_standby_led(True) +# Enable reset volume feature +speaker.set_startup_volume_enabled(True) -# Top touch panel (enable/disable) -my_speaker.get_top_panel_enabled() -my_speaker.set_top_panel_enabled(True) +# Disable reset volume (speaker resumes at last volume level) +speaker.set_startup_volume_enabled(False) +``` -# Top panel LED in active state -my_speaker.get_top_panel_led() -my_speaker.set_top_panel_led(True) +#### All Sources vs Individual Sources Mode -# Top panel LED in standby state -my_speaker.get_top_panel_standby_led() -my_speaker.set_top_panel_standby_led(False) +When reset volume is enabled, choose between "All Sources" (global) or "Individual Sources" (per-input) mode: + +```python +# Check current mode +is_all_sources = speaker.get_standby_volume_behavior() +# Returns: True = All Sources, False = Individual Sources + +# Set to "All Sources" mode (same reset volume for all inputs) +speaker.set_standby_volume_behavior(True) + +# Set to "Individual Sources" mode (different reset volume per input) +speaker.set_standby_volume_behavior(False) ``` +**How it works:** +- **All Sources (True)**: All inputs use the same reset volume (set via `defaultVolumeGlobal`) +- **Individual Sources (False)**: Each input has its own reset volume (WiFi, Bluetooth, etc.) + +#### Async Support + +All volume management methods support async: + +```python +import asyncio +import pykefcontrol as pkf + +async def manage_volumes(): + speaker = pkf.KefAsyncConnector('192.168.1.100') + + # Get all volumes + volumes = await speaker.get_all_default_volumes() + + # Set specific input volumes + await speaker.set_default_volume('wifi', 45) + await speaker.set_default_volume('bluetooth', 35) + + # Configure volume settings + await speaker.set_volume_settings(max_volume=80, step=2) + + # Enable reset volume with Individual Sources mode + await speaker.set_standby_volume_behavior(False) # Individual Sources + await speaker.set_startup_volume_enabled(True) # Enable reset volume + +asyncio.run(manage_volumes()) +``` + + +### System Behavior Settings + +Configure speaker power management, startup behavior, and inter-speaker connection settings. + +#### Auto-Standby Mode + +Control when the speaker automatically enters standby mode: + +```python +import pykefcontrol as pkf + +speaker = pkf.KefConnector('192.168.1.100') + +# Get current standby mode +mode = speaker.get_standby_mode() +print(f"Current mode: {mode}") # Returns 'standby_20mins' + +# Set standby mode +speaker.set_standby_mode('standby_20mins') # ECO mode (20 minutes) +speaker.set_standby_mode('standby_30mins') # 30 minutes +speaker.set_standby_mode('standby_60mins') # 60 minutes +speaker.set_standby_mode('standby_none') # Never auto-standby +``` + +**Standby Modes:** +- `standby_20mins` - ECO mode (shown as "ECO" in KEF Connect app) +- `standby_30mins` - 30 minutes auto-standby +- `standby_60mins` - 60 minutes auto-standby +- `standby_none` - Never auto-standby (manual standby only) + +#### Wake Source & HDMI Auto-Switch + +Configure which input wakes the speaker and HDMI auto-switching: + +```python +# Set wake source (which input can wake speaker from standby) +speaker.set_wake_source('wakeup_default') # All inputs can wake +speaker.set_wake_source('tv') # Only TV/HDMI wakes +speaker.set_wake_source('optical') # Only optical wakes + +# Enable auto-switch to HDMI when signal detected +speaker.set_auto_switch_hdmi(True) # Auto-switch enabled +speaker.set_auto_switch_hdmi(False) # Manual input selection + +# Check current settings +wake = speaker.get_wake_source() +auto_hdmi = speaker.get_auto_switch_hdmi() +print(f"Wake source: {wake}, Auto-HDMI: {auto_hdmi}") +``` + +#### Startup Behavior + +Control startup tone and USB charging: + +```python +# Disable startup beep +speaker.set_startup_tone(False) + +# Enable USB port charging +speaker.set_usb_charging(True) + +# Check current settings +tone = speaker.get_startup_tone() +usb = speaker.get_usb_charging() +``` + +#### Inter-Speaker Connection + +Configure wired vs wireless connection between left/right speakers: + +```python +# Set cable mode (for stereo pairs) +speaker.set_cable_mode('wired') # Use cable connection +speaker.set_cable_mode('wireless') # Use wireless connection + +# Set master channel designation +speaker.set_master_channel('left') # This is the left speaker +speaker.set_master_channel('right') # This is the right speaker + +# Get current settings +cable = speaker.get_cable_mode() +channel = speaker.get_master_channel() +``` + +#### Speaker Status + +Check if the speaker is powered on or in standby: + +```python +status = speaker.get_speaker_status() +if status == 'powerOn': + print("Speaker is powered on") +elif status == 'standby': + print("Speaker is in standby mode") +``` + +#### Complete Configuration Example + +```python +import pykefcontrol as pkf + +speaker = pkf.KefConnector('192.168.1.100') + +# Configure for home theater use +speaker.set_standby_mode('standby_60mins') # Long timeout +speaker.set_wake_source('tv') # Wake on TV signal +speaker.set_auto_switch_hdmi(True) # Auto-switch to HDMI +speaker.set_startup_tone(False) # Silent startup + +# Configure stereo pair +speaker.set_cable_mode('wired') # Use cable for better quality +speaker.set_master_channel('left') # Designate as left speaker +``` + +#### Async System Behavior + +All system behavior methods support async: + +```python +import asyncio +import pykefcontrol as pkf + +async def configure_speaker(): + speaker = pkf.KefAsyncConnector('192.168.1.100') + + # Get all settings + mode = await speaker.get_standby_mode() + wake = await speaker.get_wake_source() + status = await speaker.get_speaker_status() + + print(f"Standby: {mode}, Wake: {wake}, Status: {status}") + + # Configure settings + await speaker.set_standby_mode('standby_30mins') + await speaker.set_startup_tone(False) + +asyncio.run(configure_speaker()) +``` + + +### Do Not Disturb Settings + +Control LED indicators and startup behavior to minimize distractions. In the KEF Connect app, these appear under "Do Not Disturb" settings. + +**Important Note:** The API endpoints work on all KEF W2 platform speakers, but the physical effects vary by model: +- **LSX II / LSX II LT / LS50 W2 / LS60**: Only standby LED and startup tone are exposed in KEF Connect app +- **XIO Soundbar**: Full control panel LED controls (4 settings: control panel LED, control panel in standby, startup tone, control panel lock) + +#### Standby LED + +Control whether the LED indicator is visible when the speaker is in standby mode: + +```python +import pykefcontrol as pkf + +speaker = pkf.KefConnector('192.168.1.100') + +# Enable standby LED (default) +speaker.set_standby_led(True) + +# Disable standby LED (for dark rooms) +speaker.set_standby_led(False) + +# Check current setting +enabled = speaker.get_standby_led() +print(f"Standby LED: {'On' if enabled else 'Off'}") +``` + +**Async version:** +```python +enabled = await speaker.get_standby_led() +await speaker.set_standby_led(False) +``` + +#### Startup Tone + +Control the audible beep when powering on (also in System Behavior Settings): + +```python +# Disable startup beep for silent power-on +speaker.set_startup_tone(False) + +# Enable startup beep +speaker.set_startup_tone(True) + +# Check current setting +enabled = speaker.get_startup_tone() +``` + +#### XIO Soundbar: Control Panel LED Controls + +The XIO soundbar has exclusive control panel LED settings (4 controls shown in KEF Connect app under "Do Not Disturb"). The `set_top_panel_*` methods only work on XIO models. + +> **Note:** The `get/set_front_led()` methods exist for all models but have no visible effect on any currently tested speakers (LSX II, LSX II LT, XIO). The API field exists in firmware but appears to have no hardware implementation. These methods are kept for completeness in case future models support this feature. + +```python +# Control panel LED during operation +speaker.set_front_led(True) # LED on during operation (default) +speaker.set_front_led(False) # LED off during operation + +# Control panel LED in standby +speaker.set_top_panel_standby_led(True) # LED on in standby +speaker.set_top_panel_standby_led(False) # LED off in standby + +# Enable/disable top panel entirely (control panel lock) +speaker.set_top_panel_enabled(True) # Panel active (default) +speaker.set_top_panel_enabled(False) # Panel locked/disabled + +# Check current settings +front_led = speaker.get_front_led() +panel_enabled = speaker.get_top_panel_enabled() +standby_led = speaker.get_top_panel_standby_led() + +print(f"Front LED: {front_led}, Panel enabled: {panel_enabled}, Standby LED: {standby_led}") +``` + +**XIO Async version:** +```python +# XIO-specific async methods +await speaker.set_front_led(False) +await speaker.set_top_panel_standby_led(False) +await speaker.set_top_panel_enabled(False) +``` + +#### Complete Do Not Disturb Configuration + +```python +import pykefcontrol as pkf + +# Configure LSX II for bedroom use (minimal LEDs) +lsx_speaker = pkf.KefConnector('192.168.1.100') # LSX II +lsx_speaker.set_standby_led(False) # No standby indicator +lsx_speaker.set_startup_tone(False) # Silent power-on + +# Configure XIO for home theater (all LEDs off) +xio_speaker = pkf.KefConnector('192.168.1.101') # XIO Soundbar +xio_speaker.set_standby_led(False) # No standby LED +xio_speaker.set_startup_tone(False) # Silent power-on +xio_speaker.set_front_led(False) # Control panel off during operation +xio_speaker.set_top_panel_standby_led(False) # Control panel off in standby +xio_speaker.set_top_panel_enabled(False) # Lock control panel (optional) +``` + + + #### Advanced features This function is used internally by pykefcontrol and returns a JSON output with a lot of information. You might want to use them to get extra information such as the artwork/album cover URL, which does not have a dedicated function _yet_ in pykefcontrol. From bcd2458ee98fda9309b59b7cf821ac6533c0c4b4 Mon Sep 17 00:00:00 2001 From: danielpetrovic Date: Fri, 24 Apr 2026 23:09:20 +0200 Subject: [PATCH 4/6] Add LSXII fw3.0 POST support, LSX2 alias, and raise_for_status MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add "LSXII" to _POST_MODELS so fw3.0 setData calls use POST instead of GET (which returns 405 on fw3.0). Supersedes PR #16. - Add "LSX2" → "LSXII" to _MODEL_ALIASES so older firmware or API responses using the legacy name still resolve correctly - Add raise_for_status() before every response.json() call so HTTP errors surface instead of being silently swallowed, based on leccelecce's work in N0ciple/pykefcontrol#15 Co-Authored-By: Claude Sonnet 4.6 --- pykefcontrol/kef_connector.py | 110 ++++++++++++++++++++++++++++++++++ 1 file changed, 110 insertions(+) diff --git a/pykefcontrol/kef_connector.py b/pykefcontrol/kef_connector.py index 06e72f3..2aaede0 100644 --- a/pykefcontrol/kef_connector.py +++ b/pykefcontrol/kef_connector.py @@ -60,6 +60,7 @@ def _set_data(self, payload): with requests.post( "http://" + self.host + "/api/setData", json=payload ) as response: + response.raise_for_status() return response.json() else: payload = dict(payload) @@ -67,6 +68,7 @@ def _set_data(self, payload): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() return response.json() def _track_control(self, command): @@ -125,6 +127,7 @@ def get_default_volume(self, input_source): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0]["i32_"] @@ -171,6 +174,7 @@ def set_default_volume(self, input_source, volume): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_all_default_volumes(self): @@ -211,6 +215,7 @@ def get_all_default_volumes(self): "http://" + self.host + "/api/getData", params=payload ) as response: if response.status_code == 200: + response.raise_for_status() json_output = response.json() volumes[input_source] = json_output[0]["i32_"] except: @@ -236,6 +241,7 @@ def get_volume_settings(self): payload = {"path": "settings:/kef/host/maximumVolume", "roles": "value"} with requests.get("http://" + self.host + "/api/getData", params=payload) as response: if response.status_code == 200: + response.raise_for_status() settings['max_volume'] = response.json()[0]["i32_"] except: pass @@ -245,6 +251,7 @@ def get_volume_settings(self): payload = {"path": "settings:/kef/host/volumeStep", "roles": "value"} with requests.get("http://" + self.host + "/api/getData", params=payload) as response: if response.status_code == 200: + response.raise_for_status() settings['step'] = response.json()[0]["i16_"] except: pass @@ -254,6 +261,7 @@ def get_volume_settings(self): payload = {"path": "settings:/kef/host/volumeLimit", "roles": "value"} with requests.get("http://" + self.host + "/api/getData", params=payload) as response: if response.status_code == 200: + response.raise_for_status() settings['limit_enabled'] = response.json()[0]["bool_"] except: pass @@ -263,6 +271,7 @@ def get_volume_settings(self): payload = {"path": "settings:/kef/host/volumeDisplay", "roles": "value"} with requests.get("http://" + self.host + "/api/getData", params=payload) as response: if response.status_code == 200: + response.raise_for_status() settings['display'] = response.json()[0]["string_"] except: pass @@ -329,6 +338,7 @@ def get_standby_volume_behavior(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() # advancedStandbyDefaultVol: false = global, true = per-input @@ -354,6 +364,7 @@ def set_standby_volume_behavior(self, use_global): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_startup_volume_enabled(self): @@ -376,6 +387,7 @@ def get_startup_volume_enabled(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0]["bool_"] @@ -402,6 +414,7 @@ def set_startup_volume_enabled(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() # Network Diagnostics Methods (Phase 4) @@ -417,6 +430,7 @@ def _get_player_data(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0] @@ -500,6 +514,7 @@ def get_request(self, path, roles="value"): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output @@ -568,6 +583,7 @@ def _get_polling_queue(self, song_status=False, poll_song_status=False): with requests.post( "http://" + self.host + "/api/event/modifyQueue", json=payload ) as response: + response.raise_for_status() json_output = response.json() # Update polling_queue property with queue uuid @@ -640,6 +656,7 @@ def poll_speaker(self, timeout=10, song_status=False, poll_song_status=False): params=payload, timeout=timeout + 0.5, # add 0.5 seconds to timeout to allow for processing ) as response: + response.raise_for_status() json_output = response.json() # Process all events @@ -666,6 +683,7 @@ def mac_address(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0]["string_"] @@ -678,6 +696,7 @@ def speaker_name(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0]["string_"] @@ -690,6 +709,7 @@ def status(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0]["kefSpeakerStatus"] @@ -718,6 +738,7 @@ def source(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0]["kefPhysicalSource"] @@ -749,6 +770,7 @@ def volume(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0]["i32_"] @@ -793,6 +815,7 @@ def song_status(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0]["i64_"] @@ -810,6 +833,7 @@ def _get_speaker_firmware_version(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0]["string_"] @@ -848,6 +872,7 @@ def get_auto_switch_hdmi(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0].get("bool_", False) @@ -870,6 +895,7 @@ def set_auto_switch_hdmi(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_standby_mode(self): @@ -889,6 +915,7 @@ def get_standby_mode(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0].get("string_", "standby_20mins") @@ -917,6 +944,7 @@ def set_standby_mode(self, mode): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_startup_tone(self): @@ -936,6 +964,7 @@ def get_startup_tone(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0].get("bool_", False) @@ -958,6 +987,7 @@ def set_startup_tone(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_subwoofer_wake_on_startup(self): @@ -977,6 +1007,7 @@ def get_subwoofer_wake_on_startup(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0].get("bool_", False) @@ -1002,6 +1033,7 @@ def set_subwoofer_wake_on_startup(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_kw1_wake_on_startup(self): @@ -1022,6 +1054,7 @@ def get_kw1_wake_on_startup(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0].get("bool_", False) @@ -1048,6 +1081,7 @@ def set_kw1_wake_on_startup(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_wake_source(self): @@ -1067,6 +1101,7 @@ def get_wake_source(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0].get("kefWakeUpSource", "wakeup_default") @@ -1093,6 +1128,7 @@ def set_wake_source(self, source): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_usb_charging(self): @@ -1112,6 +1148,7 @@ def get_usb_charging(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0].get("bool_", False) @@ -1134,6 +1171,7 @@ def set_usb_charging(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_cable_mode(self): @@ -1153,6 +1191,7 @@ def get_cable_mode(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0].get("string_", "wired") @@ -1179,6 +1218,7 @@ def set_cable_mode(self, mode): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_master_channel(self): @@ -1198,6 +1238,7 @@ def get_master_channel(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output[0].get("kefMasterChannelMode", "right") @@ -1224,6 +1265,7 @@ def set_master_channel(self, channel): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() @@ -1248,6 +1290,7 @@ def get_front_led(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() # Note: API uses "disable" so we invert the boolean @@ -1277,6 +1320,7 @@ def set_front_led(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_standby_led(self): @@ -1296,6 +1340,7 @@ def get_standby_led(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() # Note: API uses "disable" so we invert the boolean @@ -1321,6 +1366,7 @@ def set_standby_led(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_top_panel_enabled(self): @@ -1340,6 +1386,7 @@ def get_top_panel_enabled(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() # Note: API uses "disable" so we invert the boolean @@ -1365,6 +1412,7 @@ def set_top_panel_enabled(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_top_panel_led(self): @@ -1386,6 +1434,7 @@ def get_top_panel_led(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() # Check if response is an error (dict with 'error' key) or empty if isinstance(json_output, dict) and 'error' in json_output: @@ -1412,6 +1461,7 @@ def set_top_panel_led(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() def get_top_panel_standby_led(self): @@ -1433,6 +1483,7 @@ def get_top_panel_standby_led(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() # Check if response is an error (dict with 'error' key) or empty if isinstance(json_output, dict) and 'error' in json_output: @@ -1459,6 +1510,7 @@ def set_top_panel_standby_led(self, enabled): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() # ===== Remote Control Methods ===== @@ -1485,6 +1537,7 @@ def get_fixed_volume_mode(self): with requests.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = response.json() value = json_output[0].get("i32_", -1) return None if value < 0 else value @@ -1513,6 +1566,7 @@ def set_fixed_volume_mode(self, volume): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() @@ -1537,6 +1591,7 @@ def set_request(self, path, roles="value", value=None): with requests.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = response.json() return json_output @@ -1601,6 +1656,7 @@ async def _set_data(self, payload): async with self._session.post( "http://" + self.host + "/api/setData", json=payload ) as response: + response.raise_for_status() return await response.json() else: payload = dict(payload) @@ -1608,6 +1664,7 @@ async def _set_data(self, payload): async with self._session.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() return await response.json() async def _track_control(self, command): @@ -1629,6 +1686,7 @@ async def _get_player_data(self): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output[0] @@ -1651,6 +1709,7 @@ async def get_request(self, path, roles="value"): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output @@ -1676,6 +1735,7 @@ async def set_request(self, path, roles="value", value=None): async with self._session.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output @@ -1830,6 +1890,7 @@ async def get_polling_queue(self, song_status=False, poll_song_status=False): async with self._session.post( "http://" + self.host + "/api/event/modifyQueue", json=payload ) as response: + response.raise_for_status() json_output = await response.json() # Update polling_queue property with queue uuid @@ -1899,6 +1960,7 @@ async def poll_speaker(self, timeout=10, song_status=False, poll_song_status=Fal params=payload, timeout=10 + 0.5, # add 0.5 seconds to timeout to allow for processing ) as response: + response.raise_for_status() json_output = await response.json() # Process all events @@ -1924,6 +1986,7 @@ async def mac_address(self): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output[0]["string_"] @@ -1936,6 +1999,7 @@ async def speaker_name(self): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output[0]["string_"] @@ -1948,6 +2012,7 @@ async def status(self): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output[0]["kefSpeakerStatus"] @@ -1978,6 +2043,7 @@ async def song_status(self): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output[0]["i64_"] @@ -1994,6 +2060,7 @@ async def source(self): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output[0]["kefPhysicalSource"] @@ -2009,6 +2076,7 @@ async def volume(self): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output[0]["i32_"] @@ -2026,6 +2094,7 @@ async def _get_speaker_firmware_version(self): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output[0]["string_"] @@ -2085,6 +2154,7 @@ async def get_default_volume(self, input_source): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output[0]["i32_"] @@ -2133,6 +2203,7 @@ async def set_default_volume(self, input_source, volume): async with self._session.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() async def get_all_default_volumes(self): @@ -2175,6 +2246,7 @@ async def get_all_default_volumes(self): "http://" + self.host + "/api/getData", params=payload ) as response: if response.status == 200: + response.raise_for_status() json_output = await response.json() volumes[input_source] = json_output[0]["i32_"] except: @@ -2201,6 +2273,7 @@ async def get_volume_settings(self): payload = {"path": "settings:/kef/host/maximumVolume", "roles": "value"} async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: if response.status == 200: + response.raise_for_status() json_output = await response.json() settings['max_volume'] = json_output[0]["i32_"] except: @@ -2211,6 +2284,7 @@ async def get_volume_settings(self): payload = {"path": "settings:/kef/host/volumeStep", "roles": "value"} async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: if response.status == 200: + response.raise_for_status() json_output = await response.json() settings['step'] = json_output[0]["i16_"] except: @@ -2221,6 +2295,7 @@ async def get_volume_settings(self): payload = {"path": "settings:/kef/host/volumeLimit", "roles": "value"} async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: if response.status == 200: + response.raise_for_status() json_output = await response.json() settings['limit_enabled'] = json_output[0]["bool_"] except: @@ -2231,6 +2306,7 @@ async def get_volume_settings(self): payload = {"path": "settings:/kef/host/volumeDisplay", "roles": "value"} async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: if response.status == 200: + response.raise_for_status() json_output = await response.json() settings['display'] = json_output[0]["string_"] except: @@ -2301,6 +2377,7 @@ async def get_standby_volume_behavior(self): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() # advancedStandbyDefaultVol: false = global (All sources), true = per-input (Individual sources) @@ -2327,6 +2404,7 @@ async def set_standby_volume_behavior(self, use_global): async with self._session.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() async def get_startup_volume_enabled(self): @@ -2350,6 +2428,7 @@ async def get_startup_volume_enabled(self): async with self._session.get( "http://" + self.host + "/api/getData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() return json_output[0]["bool_"] @@ -2377,6 +2456,7 @@ async def set_startup_volume_enabled(self, enabled): async with self._session.get( "http://" + self.host + "/api/setData", params=payload ) as response: + response.raise_for_status() json_output = await response.json() # Async Network Diagnostics Methods (Phase 4) @@ -2386,6 +2466,7 @@ async def get_auto_switch_hdmi(self): payload = {"path": "settings:/kef/host/autoSwitchToHDMI", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return json_output[0].get("bool_", False) @@ -2398,6 +2479,7 @@ async def set_auto_switch_hdmi(self, enabled): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_standby_mode(self): @@ -2405,6 +2487,7 @@ async def get_standby_mode(self): payload = {"path": "settings:/kef/host/standbyMode", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return json_output[0].get("string_", "standby_20mins") @@ -2420,6 +2503,7 @@ async def set_standby_mode(self, mode): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_startup_tone(self): @@ -2427,6 +2511,7 @@ async def get_startup_tone(self): payload = {"path": "settings:/kef/host/startupTone", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return json_output[0].get("bool_", False) @@ -2439,6 +2524,7 @@ async def set_startup_tone(self, enabled): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_subwoofer_wake_on_startup(self): @@ -2453,6 +2539,7 @@ async def get_subwoofer_wake_on_startup(self): payload = {"path": "settings:/kef/host/subwooferForceOn", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return json_output[0].get("bool_", False) @@ -2472,6 +2559,7 @@ async def set_subwoofer_wake_on_startup(self, enabled): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_kw1_wake_on_startup(self): @@ -2487,6 +2575,7 @@ async def get_kw1_wake_on_startup(self): payload = {"path": "settings:/kef/host/subwooferForceOnKW1", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return json_output[0].get("bool_", False) @@ -2507,6 +2596,7 @@ async def set_kw1_wake_on_startup(self, enabled): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_wake_source(self): @@ -2514,6 +2604,7 @@ async def get_wake_source(self): payload = {"path": "settings:/kef/host/wakeUpSource", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return json_output[0].get("kefWakeUpSource", "wakeup_default") @@ -2529,6 +2620,7 @@ async def set_wake_source(self, source): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_usb_charging(self): @@ -2536,6 +2628,7 @@ async def get_usb_charging(self): payload = {"path": "settings:/kef/host/usbCharging", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return json_output[0].get("bool_", False) @@ -2548,6 +2641,7 @@ async def set_usb_charging(self, enabled): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_cable_mode(self): @@ -2555,6 +2649,7 @@ async def get_cable_mode(self): payload = {"path": "settings:/kef/host/cableMode", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return json_output[0].get("string_", "wired") @@ -2570,6 +2665,7 @@ async def set_cable_mode(self, mode): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_master_channel(self): @@ -2577,6 +2673,7 @@ async def get_master_channel(self): payload = {"path": "settings:/kef/host/masterChannelMode", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return json_output[0].get("kefMasterChannelMode", "right") @@ -2592,6 +2689,7 @@ async def set_master_channel(self, channel): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() @@ -2604,6 +2702,7 @@ async def get_front_led(self): payload = {"path": "settings:/kef/host/disableFrontLED", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return not json_output[0].get("bool_", False) @@ -2621,6 +2720,7 @@ async def set_front_led(self, enabled): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_standby_led(self): @@ -2628,6 +2728,7 @@ async def get_standby_led(self): payload = {"path": "settings:/kef/host/disableFrontStandbyLED", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return not json_output[0].get("bool_", False) @@ -2641,6 +2742,7 @@ async def set_standby_led(self, enabled): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_top_panel_enabled(self): @@ -2648,6 +2750,7 @@ async def get_top_panel_enabled(self): payload = {"path": "settings:/kef/host/disableTopPanel", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() return not json_output[0].get("bool_", False) @@ -2661,6 +2764,7 @@ async def set_top_panel_enabled(self, enabled): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_top_panel_led(self): @@ -2672,6 +2776,7 @@ async def get_top_panel_led(self): payload = {"path": "settings:/kef/host/topPanelLED", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() # Check if response is an error (dict with 'error' key) or empty if isinstance(json_output, dict) and 'error' in json_output: @@ -2689,6 +2794,7 @@ async def set_top_panel_led(self, enabled): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() async def get_top_panel_standby_led(self): @@ -2700,6 +2806,7 @@ async def get_top_panel_standby_led(self): payload = {"path": "settings:/kef/host/topPanelStandbyLED", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() # Check if response is an error (dict with 'error' key) or empty if isinstance(json_output, dict) and 'error' in json_output: @@ -2717,6 +2824,7 @@ async def set_top_panel_standby_led(self, enabled): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() # ===== Remote Control Methods (Async) ===== @@ -2738,6 +2846,7 @@ async def get_fixed_volume_mode(self): payload = {"path": "settings:/kef/host/remote/userFixedVolume", "roles": "value"} await self.resurect_session() async with self._session.get("http://" + self.host + "/api/getData", params=payload) as response: + response.raise_for_status() json_output = await response.json() value = json_output[0].get("i32_", -1) return None if value < 0 else value @@ -2764,5 +2873,6 @@ async def set_fixed_volume_mode(self, volume): } await self.resurect_session() async with self._session.get("http://" + self.host + "/api/setData", params=payload) as response: + response.raise_for_status() json_output = await response.json() From 3350eedd267f60cfa0bc5356ceed2651d8ad87f1 Mon Sep 17 00:00:00 2001 From: danielpetrovic Date: Mon, 4 May 2026 10:20:53 +0200 Subject: [PATCH 5/6] Add LS60Wireless alias to map to LS60 POST model hass-kef-connector uses "LS60Wireless" as the model key (matching the product name). Without this alias, LS60 owners using hass-kef-connector PR #17 would fall back to GET-based setData and hit 405 errors with firmware 4.0+. --- pykefcontrol/kef_connector.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pykefcontrol/kef_connector.py b/pykefcontrol/kef_connector.py index 2aaede0..f76e91a 100644 --- a/pykefcontrol/kef_connector.py +++ b/pykefcontrol/kef_connector.py @@ -7,7 +7,7 @@ _POST_MODELS = {"LS50WII", "LSXIILT", "LSXII", "LS60"} -_MODEL_ALIASES = {"LS50W2": "LS50WII", "LSX2LT": "LSXIILT", "LSX2": "LSXII"} +_MODEL_ALIASES = {"LS50W2": "LS50WII", "LSX2LT": "LSXIILT", "LSX2": "LSXII", "LS60Wireless": "LS60"} class KefConnector: From 2c63e48ed24d20d3cdb95ab8b3d8d47ec02ab26c Mon Sep 17 00:00:00 2001 From: danielpetrovic Date: Wed, 27 May 2026 19:53:58 +0200 Subject: [PATCH 6/6] Add XIO and LS60W to POST models, add LS60/LS60Wireless aliases XIO firmware 1.4.135 switched setData from GET to POST. LS60 firmware renamed the model ID from "LS60" to "LS60W" (fixes upstream issue #18). Changes: - Add "XIO" and "LS60W" to _POST_MODELS - Remove "LS60" from _POST_MODELS (no longer the active model ID) - Add "LS60": "LS60W" alias for backward compat with older LS60 firmware - Add "LS60Wireless": "LS60W" alias for integrations using the product name --- pykefcontrol/kef_connector.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pykefcontrol/kef_connector.py b/pykefcontrol/kef_connector.py index f76e91a..3633160 100644 --- a/pykefcontrol/kef_connector.py +++ b/pykefcontrol/kef_connector.py @@ -6,8 +6,8 @@ import warnings -_POST_MODELS = {"LS50WII", "LSXIILT", "LSXII", "LS60"} -_MODEL_ALIASES = {"LS50W2": "LS50WII", "LSX2LT": "LSXIILT", "LSX2": "LSXII", "LS60Wireless": "LS60"} +_POST_MODELS = {"LS50WII", "LSXIILT", "LSXII", "LS60W", "XIO"} +_MODEL_ALIASES = {"LS50W2": "LS50WII", "LSX2LT": "LSXIILT", "LSX2": "LSXII", "LS60": "LS60W", "LS60Wireless": "LS60W"} class KefConnector: