From 404959abcc6ef89df75a8722fa73987a686dcace Mon Sep 17 00:00:00 2001 From: dhartung <22015466+dhartung@users.noreply.github.com> Date: Fri, 24 Jan 2025 17:53:11 +0100 Subject: [PATCH] Fixed smart mode, added support for pure voice toggle (#21) * Removed hardcoded address * Added new switch for PureVoice mode * Simplified dictionary merge --------- Authored-by: Daniel Hartung --- .../jbl_integration/coordinator.py | 172 ++++++++---------- custom_components/jbl_integration/switch.py | 60 +++++- 2 files changed, 135 insertions(+), 97 deletions(-) diff --git a/custom_components/jbl_integration/coordinator.py b/custom_components/jbl_integration/coordinator.py index 814c960..c6d4022 100644 --- a/custom_components/jbl_integration/coordinator.py +++ b/custom_components/jbl_integration/coordinator.py @@ -1,18 +1,13 @@ """Sensor platform for JBL integration.""" import asyncio import aiohttp -import requests import json import logging import urllib3 import ssl import certifi from datetime import timedelta -from homeassistant.helpers.entity import Entity -from homeassistant.config_entries import ConfigEntry -from homeassistant.core import HomeAssistant from homeassistant.helpers.update_coordinator import DataUpdateCoordinator - from .const import DOMAIN _LOGGER = logging.getLogger(__name__) @@ -25,13 +20,15 @@ def __init__(self, address, pollingRate, hass=None, entry=None): self.address = address self.pollingRate = pollingRate self.data = {} + + ssl_context = ssl.create_default_context(cafile=certifi.where()) + ssl_context.check_hostname = False + ssl_context.verify_mode = ssl.CERT_NONE + self.sslcontext = ssl_context + if hass != None and entry != None: self._entry = entry self.hass = hass - ssl_context = ssl.create_default_context(cafile=certifi.where()) - ssl_context.check_hostname = False - ssl_context.verify_mode = ssl.CERT_NONE - self.sslcontext = ssl_context super().__init__( hass, _LOGGER, @@ -106,18 +103,15 @@ async def _send_command(self, command): _LOGGER.error("Failed to send command: %s", response.status) async def _async_update_data(self): - data1 = await self.requestInfo() - data2 = await self.getEQ() - data3 = await self.getNightMode() - data4 = await self.getRearSpeaker() - data5 = await self.getSmartMode() - - data12 = self.merge_two_dicts(data1, data2) - data123 = self.merge_two_dicts(data12, data3) - data1234 = self.merge_two_dicts(data123, data4) - data12345 = self.merge_two_dicts(data1234, data5) - - combined_data = data12345 + combined_data = { + **await self.requestInfo(), + **await self.getEQ(), + **await self.getNightMode(), + **await self.getRearSpeaker(), + **await self.getSmartMode(), + **await self.getPureVoice(), + } + # Ensure self.data is initialized to an empty dictionary if it is None if self.data is None: self.data = {} @@ -125,6 +119,31 @@ async def _async_update_data(self): self.data.update(combined_data) return combined_data + async def _getCommand(self, command): + # Disable SSL warnings + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + url = f'https://{self.address}/httpapi.asp?command={command}' + + headers = { + 'Accept-Encoding': "gzip", + } + async with aiohttp.ClientSession() as session: + try: + async with asyncio.timeout(10): + async with session.get(url, headers=headers, ssl=self.sslcontext) as response: + if response.status == 200: + response_text = await response.text() + response_json = json.loads(response_text) + _LOGGER.debug(f"%s Response text: %s", command, response_text) + return response_json + else: + _LOGGER.error(f"Failed to get %s: %s", command, response.status) + return {} + except Exception as e: + _LOGGER.error(f"Error getting %s: %s", command, str(e)) + return {} + async def getDeviceInfo(self): # Disable SSL warnings urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) @@ -382,33 +401,11 @@ async def setEQ(self, value: float, frequency): return {} async def getNightMode(self): - # Disable SSL warnings - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - - url = f'https://{self.address}/httpapi.asp?command=getPersonalListeningMode' - - headers = { - 'Accept-Encoding': "gzip", - } - async with aiohttp.ClientSession() as session: - try: - async with asyncio.timeout(10): - async with session.get(url, headers=headers, ssl=self.sslcontext) as response: - if response.status == 200: - response_text = await response.text() - response_json = json.loads(response_text) - try: - _LOGGER.debug("Nightmode Response text: %s", response_text) - return {"NightMode":response_json["status"]} - except Exception as e: - _LOGGER.debug("No Nightmode available") - return {} - else: - _LOGGER.error("Nightmode Response status: %s", response.status) - return {} - except Exception as e: - _LOGGER.error("Error getting NightMode: %s", str(e)) - return {} + response = await self._getCommand("getPersonalListeningMode") + if "status" in response: + return { "NightMode": response["status"] } + else: + return {} async def setNightMode(self, value: bool): # Disable SSL warnings @@ -437,65 +434,48 @@ async def setNightMode(self, value: bool): return {} async def getRearSpeaker(self): - # Disable SSL warnings - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - - url = f'https://{self.address}/httpapi.asp?command=getRearSpeakerStatus' - - headers = { - 'Accept-Encoding': "gzip", - } - async with aiohttp.ClientSession() as session: - try: - async with asyncio.timeout(10): - async with session.get(url, headers=headers, ssl=self.sslcontext) as response: - if response.status == 200: - response_text = await response.text() - _LOGGER.debug("Rear Speakers Response text: %s", response_text) - try: - response_json = json.loads(response_text) - return {"Rears":response_json["rears"]} - except Exception as e: - _LOGGER.debug("No Rear Speakers available") - if "Rears"in self.data: - self.data.pop('Rears', None) - return {} - else: - return {} - except Exception as e: - _LOGGER.error("Error getting Rear Speakers: %s", str(e)) - return {} + response = await self._getCommand("getRearSpeakerStatus") + if "rears" in response: + return { "Rears": response["rears"] } + else: + return {} async def getSmartMode(self): + response = await self._getCommand("getSmartMode") + if "status" in response: + return { "SmartMode": response["status"] } + else: + return {} + + async def getPureVoice(self): + response = await self._getCommand("getPureVoiceState") + if "purevoice_state" in response: + return { "PureVoice": "on" if response["purevoice_state"] == "1" else "off" } + else: + return {} + + async def setPureVoice(self, value: bool): # Disable SSL warnings urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - - url = 'https://192.148.4.66/httpapi.asp?command=getSmartMode' - + + """Fetch data from the API.""" + url = f'https://{self.address}/httpapi.asp' headers = { 'Accept-Encoding': "gzip", } + + strvalue = '1' if value else '0' + payload = 'command=setPureVoiceState&payload={"purevoice_state":"'+strvalue+'"}' + async with aiohttp.ClientSession() as session: try: async with asyncio.timeout(10): - async with session.get(url, headers=headers, ssl=self.sslcontext) as response: - if response.status == 200: - response_text = await response.text() - _LOGGER.debug("SmartMode Response text: %s", response_text) - try: - response_json = json.loads(response_text) - return {"SmartMode": response_json["status"]} - except Exception as e: - _LOGGER.debug("No SmartMode available") - return {} + async with session.post(url, headers=headers, data=payload, ssl=self.sslcontext) as response: + if response.status != 200: + _LOGGER.error("Failed to set PureVoice: %s", response.status) + return {} else: - _LOGGER.error("SmartMode Response status: %s", response.status) return {} except Exception as e: - _LOGGER.error("Error getting SmartMode: %s", str(e)) + _LOGGER.error("Error setting PureVoice: %s", str(e)) return {} - - def merge_two_dicts(self,x, y): - z = x.copy() # start with keys and values of x - z.update(y) # modifies z with keys and values of y - return z \ No newline at end of file diff --git a/custom_components/jbl_integration/switch.py b/custom_components/jbl_integration/switch.py index 047b889..2fc2e84 100644 --- a/custom_components/jbl_integration/switch.py +++ b/custom_components/jbl_integration/switch.py @@ -21,7 +21,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e entityArray.append(SmartModeSwitch(entry, coordinator)) if "NightMode" in coordinator.data: entityArray.append(NightModeSwitch(entry, coordinator)) - + if "PureVoice" in coordinator.data: + entityArray.append(PureVoiceModeSwitch(entry, coordinator)) async_add_entities(entityArray) @@ -208,6 +209,63 @@ async def async_added_to_hass(self): """When entity is added to hass.""" self.async_on_remove(self.coordinator.async_add_listener(self.async_write_ha_state)) + async def async_update(self): + """Update the sensor.""" + await self.coordinator.async_request_refresh() + + +class PureVoiceModeSwitch(SwitchEntity): + """Representation of a switch to control JBL PureVoice.""" + + def __init__(self, entry: ConfigEntry, coordinator: Coordinator): + """Initialize the switch.""" + self._entry = entry + self._is_on = False + self.coordinator = coordinator + self.entity_id = f"switch.{self.coordinator.device_info.get('name', 'jbl_integration').replace(' ', '_').lower()}_pure_voice" + + @property + def name(self): + """Return the name of the switch.""" + return "JBL Pure Voice" + + @property + def unique_id(self): + """Return a unique ID for the switch.""" + return f"jbl_pure_voice_{self._entry.entry_id}" + + @property + def is_on(self): + """Return true if switch is on.""" + return self.coordinator.data.get("PureVoice") == "on" + + @property + def icon(self): + """Return the icon to use in the frontend.""" + return "mdi:account-voice" + + @property + def device_info(self): + """Return device information about this entity.""" + return self.coordinator.device_info + + async def async_turn_on(self, **kwargs): + """Turn the switch on.""" + await self.coordinator.setPureVoice(True) + + async def async_turn_off(self, **kwargs): + """Turn the switch off.""" + await self.coordinator.setPureVoice(False) + + @property + def should_poll(self): + """No polling needed.""" + return False + + async def async_added_to_hass(self): + """When entity is added to hass.""" + self.async_on_remove(self.coordinator.async_add_listener(self.async_write_ha_state)) + async def async_update(self): """Update the sensor.""" await self.coordinator.async_request_refresh() \ No newline at end of file