From 32948a7f709130d8bc5690a99316cd341591c6a5 Mon Sep 17 00:00:00 2001 From: Ernestas Date: Tue, 30 Dec 2025 22:28:06 +0200 Subject: [PATCH 1/4] wip --- .github/workflows/hassfest.yaml | 15 + custom_components/junghome/__init__.py | 34 +- custom_components/junghome/button.py | 53 +++- custom_components/junghome/config_flow.py | 43 ++- custom_components/junghome/const.py | 2 + custom_components/junghome/coordinator.py | 314 +++++++++++------- custom_components/junghome/event.py | 267 ++++++++++------ custom_components/junghome/light.py | 371 ++++++++++++++-------- custom_components/junghome/manifest.json | 2 + custom_components/junghome/sensor.py | 98 ++++-- custom_components/junghome/switch.py | 152 ++++++--- 11 files changed, 895 insertions(+), 456 deletions(-) create mode 100644 .github/workflows/hassfest.yaml diff --git a/.github/workflows/hassfest.yaml b/.github/workflows/hassfest.yaml new file mode 100644 index 0000000..c6f8aae --- /dev/null +++ b/.github/workflows/hassfest.yaml @@ -0,0 +1,15 @@ +name: Validate with hassfest + +on: + push: + pull_request: + schedule: + - cron: "0 1 * * SUN" + workflow_dispatch: + +jobs: + validate: + runs-on: "ubuntu-latest" + steps: + - uses: "actions/checkout@v6" + - uses: home-assistant/actions/hassfest@master diff --git a/custom_components/junghome/__init__.py b/custom_components/junghome/__init__.py index 7702b1e..e9ef6f4 100644 --- a/custom_components/junghome/__init__.py +++ b/custom_components/junghome/__init__.py @@ -1,13 +1,22 @@ +""" +Junghome integration package. + +Provides integration setup and entry load/unload helpers. +""" + import logging + from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.typing import ConfigType -from .coordinator import JungHomeDataUpdateCoordinator + from .const import DOMAIN +from .coordinator import JungHomeDataUpdateCoordinator _LOGGER = logging.getLogger(__name__) -async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: + +async def async_setup(_hass: HomeAssistant, _config: ConfigType) -> bool: """Set up the Jung Home integration.""" return True @@ -31,7 +40,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: await coordinator.start() # Forward the setup to the appropriate platforms - await hass.config_entries.async_forward_entry_setups(entry, ["light", "switch", "sensor", "event"]) + await hass.config_entries.async_forward_entry_setups( + entry, + ["light", "switch", "sensor", "event"], + ) return True @@ -40,15 +52,15 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: if DOMAIN in hass.data and entry.entry_id in hass.data[DOMAIN]: del hass.data[DOMAIN][entry.entry_id] - # Unload platforms - unload_ok = await hass.config_entries.async_forward_entry_unload(entry, "light") - unload_ok = unload_ok and await hass.config_entries.async_forward_entry_unload(entry, "switch") - unload_ok = unload_ok and await hass.config_entries.async_forward_entry_unload(entry, "sensor") - unload_ok = unload_ok and await hass.config_entries.async_forward_entry_unload(entry, "event") - - return unload_ok + # Unload platforms sequentially and return combined result + return ( + await hass.config_entries.async_forward_entry_unload(entry, "light") + and await hass.config_entries.async_forward_entry_unload(entry, "switch") + and await hass.config_entries.async_forward_entry_unload(entry, "sensor") + and await hass.config_entries.async_forward_entry_unload(entry, "event") + ) async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Reload a config entry.""" await async_unload_entry(hass, entry) - return await async_setup_entry(hass, entry) \ No newline at end of file + return await async_setup_entry(hass, entry) diff --git a/custom_components/junghome/button.py b/custom_components/junghome/button.py index 04f1bd3..6663561 100644 --- a/custom_components/junghome/button.py +++ b/custom_components/junghome/button.py @@ -1,35 +1,66 @@ +""" +Button event forwarding for the Jung Home integration. + +This module listens for internal dispatcher events and forwards +stateless button presses to the Home Assistant event bus. +""" + +from __future__ import annotations + import logging +from typing import TYPE_CHECKING, Any + from homeassistant.helpers.dispatcher import async_dispatcher_connect -from .const import DOMAIN + +if TYPE_CHECKING: + from collections.abc import Callable, Mapping + + from homeassistant.config_entries import ConfigEntry + from homeassistant.core import HomeAssistant _LOGGER = logging.getLogger(__name__) -SIGNAL_JUNG_BUTTON_EVENT = "jung_home_button_event" +SIGNAL_JUNG_BUTTON_EVENT: str = "jung_home_button_event" -async def async_setup_entry(hass, entry, async_add_entities): + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + _async_add_entities: Callable[..., None] | None = None, +) -> None: """Set up Jung Home button event forwarding (no entities).""" - def handle_button_event(message): - _LOGGER.debug("[JUNGHOME] handle_button_event called with message: %s", message) + + def handle_button_event(message: Mapping[str, Any]) -> None: + """Handle a dispatcher message and fire a HA event on button press.""" + _LOGGER.debug("handle_button_event called with message: %s", message) data = message.get("data") if not data: return + device_id = data.get("id") values = data.get("values", []) + + # Look for stateless press requests for value in values: - if value["key"] in {"up_request", "down_request", "trigger_request"} and value["value"] == "1": - _LOGGER.debug("Stateless button event detected for device %s", device_id) + if ( + value.get("key") + in {"up_request", "down_request", "trigger_request"} + and value.get("value") == "1" + ): + _LOGGER.debug( + "Stateless button event detected for device %s", device_id + ) hass.bus.fire( "jung_home_button_press", { "device_id": device_id, "datapoint_type": data.get("type"), - } + }, ) break + entry.async_on_unload( async_dispatcher_connect(hass, SIGNAL_JUNG_BUTTON_EVENT, handle_button_event) ) - # No entities to add - return -# Remove button.py as event.py now handles all button logic +# No entities to add; event forwarding only. diff --git a/custom_components/junghome/config_flow.py b/custom_components/junghome/config_flow.py index e6e70a6..c33345f 100644 --- a/custom_components/junghome/config_flow.py +++ b/custom_components/junghome/config_flow.py @@ -1,13 +1,23 @@ -from homeassistant import config_entries +"""Config flow for the Junghome integration.""" + +from __future__ import annotations + +from typing import Any + import voluptuous as vol -from .const import DOMAIN # Define DOMAIN in const.py (e.g., DOMAIN = "junghome") -from homeassistant.core import HomeAssistant +from homeassistant import config_entries + +from .const import DOMAIN + +# Expected token length used for simple validation +TOKEN_EXPECTED_LENGTH = 95 +TOKEN_FIELD = "token" # Example validation schema STEP_USER_DATA_SCHEMA = vol.Schema( { vol.Required("host"): str, - vol.Required("token"): str, + vol.Required(TOKEN_FIELD): str, } ) @@ -17,24 +27,29 @@ class JungHomeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): VERSION = 1 CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> config_entries.FlowResult: """Handle the initial step.""" - errors = {} + errors: dict[str, str] = {} if user_input is not None: # Validate user input here host = user_input["host"] - token = user_input["token"] + token_value = user_input[TOKEN_FIELD] if not self._validate_host(host): errors["host"] = "invalid_host" - if not self._validate_token(token): - errors["token"] = "invalid_token" + if not self._validate_token(token_value): + errors[TOKEN_FIELD] = "invalid_token" if not errors: # Save configuration if validation passes - return self.async_create_entry(title="Jung Home", data=user_input) + return self.async_create_entry( + title="Jung Home", + data={"host": host, TOKEN_FIELD: token_value}, + ) return self.async_show_form( step_id="user", @@ -42,12 +57,12 @@ async def async_step_user(self, user_input=None): errors=errors, ) - def _validate_host(self, host): + def _validate_host(self, host: str) -> bool: """Validate the host (e.g., check format or connectivity).""" # Example: Ensure the host is a valid IP address or hostname return isinstance(host, str) and len(host) > 0 - def _validate_token(self, token): + def _validate_token(self, token: str) -> bool: """Validate the API token.""" - # Example: Ensure the token has a minimum length - return isinstance(token, str) and len(token) == 95 + # Example: Check token is a string and matches expected length. + return isinstance(token, str) and len(token) == TOKEN_EXPECTED_LENGTH diff --git a/custom_components/junghome/const.py b/custom_components/junghome/const.py index 01496de..640c38a 100644 --- a/custom_components/junghome/const.py +++ b/custom_components/junghome/const.py @@ -1 +1,3 @@ +"""Integration constants for the Junghome integration.""" + DOMAIN = "junghome" diff --git a/custom_components/junghome/coordinator.py b/custom_components/junghome/coordinator.py index 37e4d35..1830a99 100644 --- a/custom_components/junghome/coordinator.py +++ b/custom_components/junghome/coordinator.py @@ -1,177 +1,250 @@ +""" +Coordinator for the Junghome integration. + +Handles polling the REST API and maintaining a websocket connection for +real-time updates. +""" + +from __future__ import annotations + import asyncio import json import logging import ssl -import aiohttp -import websockets from datetime import timedelta -from homeassistant.core import HomeAssistant +from typing import TYPE_CHECKING, Any + +import aiohttp +from homeassistant.helpers.dispatcher import async_dispatcher_send from homeassistant.helpers.update_coordinator import DataUpdateCoordinator +if TYPE_CHECKING: + from homeassistant.core import HomeAssistant + _LOGGER = logging.getLogger(__name__) class JungHomeDataUpdateCoordinator(DataUpdateCoordinator): """Class to manage fetching data from the Jung Home API.""" - def __init__(self, hass: HomeAssistant, config: dict): + def __init__(self, hass: HomeAssistant, config: dict[str, Any]) -> None: """Initialize the coordinator.""" self.hass = hass self.config = config - self.websocket = None - super().__init__(hass, _LOGGER, name="Jung Home", update_interval=timedelta(minutes=1)) + self.websocket: Any = None + super().__init__( + hass, + _LOGGER, + name="Jung Home", + update_interval=timedelta(minutes=1), + ) - async def _async_update_data(self): + async def _async_update_data(self) -> list[dict[str, Any]]: """Fetch data from the API.""" _LOGGER.debug("Fetching new device data from Jung Home API") try: - response = await self._fetch_devices_from_api(self.config['host'], self.config['token']) - if response is None: - _LOGGER.error("Received None response from API") - return [] # Returning empty list ensures entities don't break - - _LOGGER.debug("API Response: %s", response) - return response # `async_set_updated_data` is automatically called with this - except Exception as e: - _LOGGER.error("Error fetching data from Jung Home API: %s", e) - raise # Raising exception allows Home Assistant to handle errors properly - - async def _fetch_devices_from_api(self, host, token): + response = await self._fetch_devices_from_api( + self.config["host"], self.config["token"] + ) + except Exception: + _LOGGER.exception("Error fetching data from Jung Home API") + raise + + if response is None: + _LOGGER.exception("Received None response from API") + return [] # Returning empty list ensures entities don't break + + _LOGGER.debug("API Response: %s", response) + return response + + async def _fetch_devices_from_api( + self, host: str, token: str + ) -> list[dict[str, Any]]: """Fetch devices from the Jung Home API.""" ssl_context = await asyncio.to_thread(ssl.create_default_context) ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE url = f"https://{host}/api/junghome/functions" - headers = { - "token": f"{token}", - "Content-Type": "application/json" - } + headers = {"token": f"{token}", "Content-Type": "application/json"} - async with aiohttp.ClientSession() as session: - async with session.get(url, headers=headers, ssl=ssl_context) as response: - response.raise_for_status() - data = await response.json() + async with aiohttp.ClientSession() as session, session.get( + url, headers=headers, ssl=ssl_context + ) as response: + response.raise_for_status() + data = await response.json() - devices = [] - for device in data: - devices.append({ + return [ + { "id": device["id"], "label": device["label"], "type": device["type"], - "datapoints": device["datapoints"] - }) - - return devices + "datapoints": device["datapoints"], + } + for device in data + ] - async def _connect_websocket(self): + async def _connect_websocket(self) -> None: """Connect to the WebSocket and handle incoming messages using aiohttp.""" url = f"wss://{self.config['host']}/ws" - headers = { - "token": f"{self.config['token']}" - } + headers = {"token": f"{self.config['token']}"} ssl_context = ssl.create_default_context() ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE try: - async with aiohttp.ClientSession() as session: - async with session.ws_connect(url, headers=headers, ssl=ssl_context) as ws: - self.websocket = ws - _LOGGER.debug("WebSocket connected (aiohttp)") - async for msg in ws: - if msg.type == aiohttp.WSMsgType.TEXT: - _LOGGER.debug("Received WebSocket message: %s", msg.data) - try: - data = json.loads(msg.data) - if isinstance(data, list): - _LOGGER.error("Received WebSocket message is a list: %s", data) - continue - if data.get("type") in ["message", "version"]: - _LOGGER.debug("Received initial message: %s", data) - continue - self._handle_websocket_message(data) - except json.JSONDecodeError as e: - _LOGGER.error("Error decoding WebSocket message: %s", e) - except Exception as e: - _LOGGER.error("Unexpected error handling WebSocket message: %s", e) - _LOGGER.error("Message content: %s", msg.data) - elif msg.type == aiohttp.WSMsgType.ERROR: - _LOGGER.error("WebSocket error: %s", msg) - break - except Exception as e: - _LOGGER.error("Error connecting to WebSocket (aiohttp): %s", e) + async with aiohttp.ClientSession() as session, session.ws_connect( + url, headers=headers, ssl=ssl_context + ) as ws: + self.websocket = ws + _LOGGER.debug("WebSocket connected (aiohttp)") + async for msg in ws: + if msg.type == aiohttp.WSMsgType.TEXT: + _LOGGER.debug("Received WebSocket message: %s", msg.data) + try: + data = json.loads(msg.data) + if isinstance(data, list): + _LOGGER.exception( + "Received WebSocket message is a list: %s", data + ) + continue + if data.get("type") in ["message", "version"]: + _LOGGER.debug("Received initial message: %s", data) + continue + self._handle_websocket_message(data) + except json.JSONDecodeError: + _LOGGER.exception("Error decoding WebSocket message") + except Exception: + _LOGGER.exception( + "Unexpected error handling WebSocket message" + ) + _LOGGER.exception("Message content: %s", msg.data) + elif msg.type == aiohttp.WSMsgType.ERROR: + _LOGGER.exception("WebSocket error: %s", msg) + break + except (aiohttp.ClientError, TimeoutError): + _LOGGER.exception("Error connecting to WebSocket (aiohttp)") self.websocket = None - def _handle_websocket_message(self, message): + def _handle_websocket_message(self, message: Any) -> None: """Handle incoming WebSocket messages.""" if not isinstance(message, dict): - _LOGGER.error("Received WebSocket message is not a dictionary: %s", message) + _LOGGER.exception( + "Received WebSocket message is not a dictionary: %s", message + ) return - data = message.get("data") msg_type = message.get("type") - # Fire dispatcher signal for datapoint messages + + # Fire dispatcher signal for datapoint messages and call an optional + # registered callback. Keep this lightweight and delegate heavier + # processing to a separate helper to reduce branching complexity. if msg_type == "datapoint": - _LOGGER.debug("[JUNGHOME] Dispatching button event for message: %s", message) - from homeassistant.helpers.dispatcher import async_dispatcher_send + _LOGGER.debug( + "[JUNGHOME] Dispatching button event for message: %s", + message, + ) async_dispatcher_send(self.hass, "jung_home_button_event", message) - # Call button callback for datapoint messages if registered - if msg_type == "datapoint" and hasattr(self, "_button_callback"): - self._button_callback(message) - # ...existing code for dict/list handling... + if hasattr(self, "_button_callback"): + self._button_callback(message) + + # Delegate processing of the data payload to a helper. + self._process_websocket_data(data, msg_type, message) + + def _process_websocket_data( + self, + data: Any, + msg_type: Any, + raw_message: Any, + ) -> None: + """ + Process the data payload from a WebSocket message. + + This helper keeps the main handler smaller and reduces cyclomatic + complexity for linting. + """ if isinstance(data, dict): - datapoint_id = data.get("id") - if not datapoint_id: - _LOGGER.error("Received WebSocket message without datapoint_id: %s", message) - return - updated = False - for device in self.data: - for datapoint in device["datapoints"]: - if datapoint["id"] == datapoint_id: - # Update all keys in the datapoint with the new data - for key, value in data.items(): - if key != "id": - datapoint[key] = value - _LOGGER.debug("Updated datapoint for device %s: %s", device["id"], datapoint) - updated = True - break - if updated: + self._handle_datapoint_dict(data, raw_message) + return + + if isinstance(data, list): + self._handle_data_list(data, msg_type) + return + + _LOGGER.warning( + "Received WebSocket message with unknown data type: %s", + raw_message, + ) + + def _handle_datapoint_dict(self, data: dict[str, Any], raw_message: Any) -> None: + """Handle a datapoint update represented as a dictionary.""" + datapoint_id = data.get("id") + if not datapoint_id: + _LOGGER.exception( + "Received WebSocket message without datapoint_id: %s", raw_message + ) + return + + updated = False + for device in self.data: + for datapoint in device["datapoints"]: + if datapoint["id"] == datapoint_id: + # Update all keys in the datapoint with the new data + for key, value in data.items(): + if key != "id": + datapoint[key] = value + _LOGGER.debug( + "Updated datapoint for device %s: %s", + device["id"], + datapoint, + ) + updated = True break if updated: - self.async_set_updated_data(self.data) - else: - _LOGGER.warning("No matching datapoint found for id %s", datapoint_id) - elif isinstance(data, list): - if msg_type == "groups": - self.groups = data - _LOGGER.debug("Updated groups: %s", data) - elif msg_type == "scenes": - self.scenes = data - _LOGGER.debug("Updated scenes: %s", data) + break + + if updated: self.async_set_updated_data(self.data) - elif not isinstance(data, dict): - _LOGGER.warning("Received WebSocket message with unknown data type: %s", message) + else: + _LOGGER.warning("No matching datapoint found for id %s", datapoint_id) - async def start(self): - """Start the coordinator by fetching initial data and connecting to the WebSocket.""" - _LOGGER.debug("Starting coordinator: fetching initial data and connecting to WebSocket") + def _handle_data_list(self, data: list[Any], msg_type: Any) -> None: + """Handle WebSocket messages where `data` is a list (groups/scenes).""" + if msg_type == "groups": + self.groups = data + _LOGGER.debug("Updated groups: %s", data) + elif msg_type == "scenes": + self.scenes = data + _LOGGER.debug("Updated scenes: %s", data) + self.async_set_updated_data(self.data) + + async def start(self) -> None: + """ + Start the coordinator. + + Fetch initial data and connect to the WebSocket. + """ + _LOGGER.debug( + "Starting coordinator: fetching initial data and connecting to WebSocket" + ) await self.async_refresh() self.hass.loop.create_task(self._connect_websocket()) - async def send_websocket_message(self, message): + async def send_websocket_message(self, message: Any) -> None: """Send a message via WebSocket.""" _LOGGER.debug("Sending WebSocket message: %s", message) - if self.websocket and not self.websocket.closed: + if self.websocket and not getattr(self.websocket, "closed", False): try: await self.websocket.send_str(json.dumps(message)) _LOGGER.debug("WebSocket message sent successfully") - except Exception as e: - _LOGGER.error("Error sending WebSocket message: %s", e) + except (aiohttp.ClientError, TimeoutError): + _LOGGER.exception("Error sending WebSocket message") else: - _LOGGER.error("WebSocket is not connected or is closed. Attempting to reconnect...") + _LOGGER.exception( + "WebSocket is not connected or is closed. Attempting to reconnect..." + ) # Try to reconnect self.hass.loop.create_task(self._connect_websocket()) - async def turn_on_switch(self, datapoint_id): + async def turn_on_switch(self, datapoint_id: str) -> None: """Turn on the switch.""" _LOGGER.debug("Turning on switch with datapoint_id: %s", datapoint_id) message = { @@ -184,7 +257,8 @@ async def turn_on_switch(self, datapoint_id): } await self.send_websocket_message(message) - async def turn_off_switch(self, datapoint_id): + + async def turn_off_switch(self, datapoint_id: str) -> None: """Turn off the switch.""" _LOGGER.debug("Turning off switch with datapoint_id: %s", datapoint_id) message = { @@ -197,7 +271,7 @@ async def turn_off_switch(self, datapoint_id): } await self.send_websocket_message(message) - async def turn_on_light(self, datapoint_id): + async def turn_on_light(self, datapoint_id: str) -> None: """Turn on the light.""" _LOGGER.debug("Turning on light with datapoint_id: %s", datapoint_id) message = { @@ -210,7 +284,7 @@ async def turn_on_light(self, datapoint_id): } await self.send_websocket_message(message) - async def turn_off_light(self, datapoint_id): + async def turn_off_light(self, datapoint_id: str) -> None: """Turn off the light.""" _LOGGER.debug("Turning off light with datapoint_id: %s", datapoint_id) message = { @@ -223,7 +297,7 @@ async def turn_off_light(self, datapoint_id): } await self.send_websocket_message(message) - async def set_brightness(self, datapoint_id, brightness): + async def set_brightness(self, datapoint_id: str, brightness: int) -> None: """Set the brightness of the light.""" message = { "type": "datapoint", @@ -235,7 +309,7 @@ async def set_brightness(self, datapoint_id, brightness): } await self.send_websocket_message(message) - async def set_color_temp(self, datapoint_id, color_temp): + async def set_color_temp(self, datapoint_id: str, color_temp: int) -> None: """Set the color temperature of the light.""" message = { "type": "datapoint", @@ -247,9 +321,9 @@ async def set_color_temp(self, datapoint_id, color_temp): } await self.send_websocket_message(message) - async def set_status_led(self, datapoint_id, state): - """Set the status LED on (True) or off (False).""" - value = "1" if state else "0" + async def set_status_led(self, datapoint_id: str, *, is_on: bool) -> None: + """Set the status LED on (`is_on=True`) or off (`is_on=False`).""" + value = "1" if is_on else "0" message = { "type": "datapoint", "data": { @@ -258,4 +332,4 @@ async def set_status_led(self, datapoint_id, state): "values": [{"key": "status_led", "value": value}] } } - await self.send_websocket_message(message) \ No newline at end of file + await self.send_websocket_message(message) diff --git a/custom_components/junghome/event.py b/custom_components/junghome/event.py index 27682b7..8db6446 100644 --- a/custom_components/junghome/event.py +++ b/custom_components/junghome/event.py @@ -1,20 +1,47 @@ -import json +""" +Button, BinarySensor and Event entities for the Jung Home integration. + +This module adapts coordinator datapoints into Home Assistant +`ButtonEntity`, `BinarySensorEntity` and `EventEntity` instances and +fires integration events on presses. +""" + +from __future__ import annotations + import logging import time -from homeassistant.components.button import ButtonEntity +from typing import TYPE_CHECKING, Any, ClassVar + from homeassistant.components.binary_sensor import BinarySensorEntity -from homeassistant.components.switch import SwitchEntity +from homeassistant.components.button import ButtonEntity +from homeassistant.components.event import EventEntity from homeassistant.core import HomeAssistant, callback -from homeassistant.config_entries import ConfigEntry from homeassistant.helpers.update_coordinator import CoordinatorEntity -from homeassistant.components.event import EventEntity from homeassistant.util import dt as dt_util + from .const import DOMAIN +if TYPE_CHECKING: + from collections.abc import Callable, Mapping + + from homeassistant.config_entries import ConfigEntry + _LOGGER = logging.getLogger(__name__) -async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities): - """Set up Jung Home button, binary_sensor, and event entities from a config entry.""" +# Double-click detection constants +DOUBLE_CLICK_WINDOW = 0.5 +DOUBLE_CLICK_COUNT = 2 + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: Callable[..., None], +) -> None: + """ + Set up Jung Home button, binary_sensor and event entities. + + Entities are created from the coordinator data for RockerSwitch devices. + """ coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"] await coordinator.async_refresh() @@ -22,11 +49,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e entities = [] for device in devices: - if device.get('type') == 'RockerSwitch': - for datapoint in device.get('datapoints', []): - dp_type = datapoint.get('type') - if dp_type in {'down_request', 'up_request', 'trigger_request'}: - entities.append(JungHomeEventEntity(coordinator, device, datapoint)) + if device.get("type") != "RockerSwitch": + continue + for datapoint in device.get("datapoints", []): + dp_type = datapoint.get("type") + if dp_type in {"down_request", "up_request", "trigger_request"}: + entities.append( + JungHomeEventEntity(coordinator, device, datapoint) + ) # Do not create JungHomeSwitch here; it will be handled in switch.py if entities: @@ -38,17 +68,24 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e class JungHomeButton(CoordinatorEntity, ButtonEntity): """Representation of a Jung Home rocker switch as a button entity.""" - def __init__(self, coordinator, device, datapoint): + def __init__( + self, + coordinator: Any, + device: Mapping[str, Any], + datapoint: Mapping[str, Any], + ) -> None: """Initialize the button.""" super().__init__(coordinator) self._device = device self._datapoint = datapoint - self._attr_name = f"{device.get('label', 'Jung Button')}_{datapoint.get('type', 'Unknown')}" + label = device.get("label", "Jung Button") + dp_type = datapoint.get("type", "Unknown") + self._attr_name = f"{label}_{dp_type}" self._attr_unique_id = f"{device.get('id')}_{datapoint.get('id')}_button" self._attr_available = coordinator.last_update_success @property - def device_info(self): + def device_info(self) -> dict[str, Any]: """Return device information for this button.""" return { "identifiers": {(DOMAIN, self._device["id"])}, @@ -58,7 +95,7 @@ def device_info(self): "sw_version": self._device.get("sw_version", "Unknown Version"), } - async def async_press(self): + async def async_press(self) -> None: """Handle the button press (fires an event).""" _LOGGER.debug("Button %s pressed", self._attr_name) self.hass.bus.fire( @@ -66,8 +103,8 @@ async def async_press(self): { "entity_id": self.entity_id, "device_id": self._device["id"], - "datapoint_id": self._datapoint["id"] - } + "datapoint_id": self._datapoint["id"], + }, ) # ------------------------------------------ @@ -76,12 +113,19 @@ async def async_press(self): class JungHomeBinarySensor(CoordinatorEntity, BinarySensorEntity): """Representation of a Jung Home button as a binary sensor (tracks press state).""" - def __init__(self, coordinator, device, datapoint): + def __init__( + self, + coordinator: Any, + device: Mapping[str, Any], + datapoint: Mapping[str, Any], + ) -> None: """Initialize the binary sensor.""" super().__init__(coordinator) self._device = device self._datapoint = datapoint - self._attr_name = f"{device.get('label', 'Jung Button')}_{datapoint.get('type', 'Unknown')}_state" + label = device.get("label", "Jung Button") + dp_type = datapoint.get("type", "Unknown") + self._attr_name = f"{label}_{dp_type}_state" self._attr_unique_id = f"{device.get('id')}_{datapoint.get('id')}_sensor" self._attr_device_class = "occupancy" self._attr_icon = "mdi:gesture-tap-button" @@ -94,7 +138,7 @@ def __init__(self, coordinator, device, datapoint): self._last_value = self._get_state_from_datapoint(datapoint) @property - def device_info(self): + def device_info(self) -> dict[str, Any]: """Return device information for this sensor.""" return { "identifiers": {(DOMAIN, self._device["id"])}, @@ -106,76 +150,100 @@ def device_info(self): @callback def _handle_coordinator_update(self) -> None: - """Handle updated data from the coordinator (check if button is pressed or released).""" + """ + Handle updated data from the coordinator. + + Check if button is pressed or released. + """ _LOGGER.debug("Updating binary sensor for %s", self._attr_name) - device = next((d for d in self.coordinator.data if d["id"] == self._device["id"]), None) + device = next( + (d for d in self.coordinator.data if d["id"] == self._device["id"]), + None, + ) if not device: return - datapoint = next((dp for dp in device.get("datapoints", []) if dp["id"] == self._datapoint["id"]), None) + datapoint = next( + ( + dp + for dp in device.get("datapoints", []) + if dp["id"] == self._datapoint["id"] + ), + None, + ) if not datapoint: return new_state = self._get_state_from_datapoint(datapoint) - if new_state != self._last_value: - now = time.time() - if new_state is True: - self._attr_is_on = True - self.async_write_ha_state() - self.hass.bus.fire( - "jung_home_button_state_change", - { - "entity_id": self.entity_id, - "device_id": self._device["id"], - "datapoint_id": self._datapoint["id"], - "state": "pressed" - } - ) - _LOGGER.debug("Fired state change event: pressed") - # Double-click detection - if now - self._last_press_time < 0.5: - self._press_count += 1 - else: - self._press_count = 1 - self._last_press_time = now - if self._press_count == 2: - self.hass.bus.fire( - "jung_home_button_double_press", - { - "entity_id": self.entity_id, - "device_id": self._device["id"], - "datapoint_id": self._datapoint["id"] - } - ) - _LOGGER.debug("Fired double press event for %s", self._attr_name) - self._press_count = 0 - elif new_state is False: - self._attr_is_on = False - self.async_write_ha_state() + if new_state == self._last_value: + return + + now = time.time() + if new_state is True: + self._attr_is_on = True + self.async_write_ha_state() + self.hass.bus.fire( + "jung_home_button_state_change", + { + "entity_id": self.entity_id, + "device_id": self._device["id"], + "datapoint_id": self._datapoint["id"], + "state": "pressed", + }, + ) + _LOGGER.debug("Fired state change event: pressed") + + # Double-click detection + if now - self._last_press_time < DOUBLE_CLICK_WINDOW: + self._press_count += 1 + else: + self._press_count = 1 + + self._last_press_time = now + if self._press_count == DOUBLE_CLICK_COUNT: self.hass.bus.fire( - "jung_home_button_state_change", + "jung_home_button_double_press", { "entity_id": self.entity_id, "device_id": self._device["id"], "datapoint_id": self._datapoint["id"], - "state": "released" - } + }, ) - _LOGGER.debug("Fired state change event: released") - self._last_value = new_state + _LOGGER.debug("Fired double press event for %s", self._attr_name) + self._press_count = 0 + elif new_state is False: + self._attr_is_on = False + self.async_write_ha_state() + self.hass.bus.fire( + "jung_home_button_state_change", + { + "entity_id": self.entity_id, + "device_id": self._device["id"], + "datapoint_id": self._datapoint["id"], + "state": "released", + }, + ) + _LOGGER.debug("Fired state change event: released") + + self._last_value = new_state @property - def is_on(self): + def is_on(self) -> bool: """Return True if button is pressed.""" return bool(self._attr_is_on) - def _get_state_from_datapoint(self, datapoint): - """Extract state from datapoint values. Always returns True or False.""" - for value in datapoint.get('values', []): - if value['key'] in {'up_request', 'down_request', 'trigger_request'}: - if value['value'] == '1': + def _get_state_from_datapoint(self, datapoint: Mapping[str, Any]) -> bool: + """ + Extract state from datapoint values. Always returns True or False. + + Checks the request keys for explicit '1' (pressed) or '0' (released). + """ + keys = {"up_request", "down_request", "trigger_request"} + for value in datapoint.get("values", []): + if value.get("key") in keys: + if value.get("value") == "1": return True - elif value['value'] == '0': + if value.get("value") == "0": return False return False @@ -185,14 +253,21 @@ def _get_state_from_datapoint(self, datapoint): class JungHomeEventEntity(CoordinatorEntity, EventEntity): """Event entity for Jung Home button presses.""" - _attr_event_types = ["pressed", "depressed"] + _attr_event_types: ClassVar[list[str]] = ["pressed", "depressed"] - def __init__(self, coordinator, device, datapoint): + def __init__( + self, + coordinator: Any, + device: Mapping[str, Any], + datapoint: Mapping[str, Any], + ) -> None: """Initialize the event entity.""" super().__init__(coordinator) self._device = device self._datapoint = datapoint - self._attr_name = f"{device.get('label', 'Jung Button')}_{datapoint.get('type', 'Unknown')}_event" + label = device.get("label", "Jung Button") + dp_type = datapoint.get("type", "Unknown") + self._attr_name = f"{label}_{dp_type}_event" self._attr_unique_id = f"{device.get('id')}_{datapoint.get('id')}_event" self._attr_icon = "mdi:gesture-tap-button" self._attr_available = True @@ -200,7 +275,7 @@ def __init__(self, coordinator, device, datapoint): self._last_value = self._get_state_from_datapoint(datapoint) @property - def device_info(self): + def device_info(self) -> dict[str, Any]: """Return device information for this event entity.""" return { "identifiers": {(DOMAIN, self._device["id"])}, @@ -213,39 +288,53 @@ def device_info(self): @callback def _handle_coordinator_update(self) -> None: """Handle updated data from the coordinator (trigger event on press).""" - device = next((d for d in self.coordinator.data if d["id"] == self._device["id"]), None) + device = next( + (d for d in self.coordinator.data if d["id"] == self._device["id"]), + None, + ) if not device: return - datapoint = next((dp for dp in device.get("datapoints", []) if dp["id"] == self._datapoint["id"]), None) + datapoint = next( + ( + dp + for dp in device.get("datapoints", []) + if dp["id"] == self._datapoint["id"] + ), + None, + ) if not datapoint: return new_state = self._get_state_from_datapoint(datapoint) if new_state != self._last_value: now = time.time() if new_state is True: - _LOGGER.debug(f"Triggering event for {self._attr_name} at {now}") + _LOGGER.debug("Triggering event for %s at %s", self._attr_name, now) self._trigger_event("pressed") self._attr_event_timestamp = dt_util.now() self.async_write_ha_state() - _LOGGER.debug(f"EventEntity state after trigger: {self.state}") + _LOGGER.debug("EventEntity state after trigger: %s", self.state) self._last_press_time = now elif new_state is False: - _LOGGER.debug(f"Triggering depressed event for {self._attr_name} at {now}") + _LOGGER.debug( + "Triggering depressed event for %s at %s", + self._attr_name, + now, + ) self._trigger_event("depressed") self._attr_event_timestamp = dt_util.now() self.async_write_ha_state() - _LOGGER.debug(f"EventEntity state after trigger: {self.state}") + _LOGGER.debug("EventEntity state after trigger: %s", self.state) self._last_value = new_state @property - def state(self): - # Show the last event timestamp as state if available - return getattr(self, '_attr_event_timestamp', None) + def state(self) -> Any | None: + """Return the last event timestamp, or None if not available.""" + return getattr(self, "_attr_event_timestamp", None) - def _get_state_from_datapoint(self, datapoint): + def _get_state_from_datapoint(self, datapoint: Mapping[str, Any]) -> bool: """Extract state from datapoint values. Returns True if pressed.""" - for value in datapoint.get('values', []): - if value['key'] in {'up_request', 'down_request', 'trigger_request'}: - if value['value'] == '1': - return True + keys = {"up_request", "down_request", "trigger_request"} + for value in datapoint.get("values", []): + if value.get("key") in keys and value.get("value") == "1": + return True return False diff --git a/custom_components/junghome/light.py b/custom_components/junghome/light.py index b5a0838..072bf5b 100644 --- a/custom_components/junghome/light.py +++ b/custom_components/junghome/light.py @@ -1,12 +1,23 @@ -import json +"""Light entities for the Junghome integration.""" + +from __future__ import annotations + import logging import time -from homeassistant.components.light import LightEntity, ColorMode -from homeassistant.helpers.update_coordinator import CoordinatorEntity -from homeassistant.config_entries import ConfigEntry +from typing import TYPE_CHECKING, Any + +from homeassistant.components.light import ColorMode, LightEntity from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.update_coordinator import CoordinatorEntity + from .const import DOMAIN -from .coordinator import JungHomeDataUpdateCoordinator + +if TYPE_CHECKING: + from collections.abc import Callable, Mapping + + from homeassistant.config_entries import ConfigEntry + + from .coordinator import JungHomeDataUpdateCoordinator _LOGGER = logging.getLogger(__name__) @@ -14,18 +25,34 @@ DEFAULT_MAX_KELVIN = 6500 def kelvin_to_mired(kelvin: int) -> int: + """ + Convert Kelvin to mireds, with safe fallback. + + Returns a rounded integer mired value. Falls back to the default + minimum kelvin on invalid input. + """ try: return round(1000000 / kelvin) - except Exception: + except (TypeError, ZeroDivisionError): return round(1000000 / DEFAULT_MIN_KELVIN) def mired_to_kelvin(mired: int) -> int: + """ + Convert mireds to Kelvin, with safe fallback. + + Returns a rounded integer Kelvin value. Falls back to the default + minimum kelvin on invalid input. + """ try: return round(1000000 / mired) - except Exception: + except (TypeError, ZeroDivisionError): return DEFAULT_MIN_KELVIN -async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: Callable, +) -> None: """Set up Jung Home lights from a config entry.""" coordinator = hass.data[DOMAIN][config_entry.entry_id]["coordinator"] @@ -34,13 +61,13 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry, asyn devices = coordinator.data # Create entities for each light device - entities = [] - for device in devices: - _LOGGER.debug("Processing device: %s", device) - if device['type'] == 'OnOff' or device['type'] == 'ColorLight': # Add devices with type "OnOff" or "ColorLight" - for datapoint in device.get('datapoints', []): - if datapoint['type'] == 'switch': - entities.append(JungHomeLight(coordinator, device, datapoint)) + entities = [ + JungHomeLight(coordinator, device, datapoint) + for device in devices + for datapoint in device.get("datapoints", []) + if device.get("type") in ("OnOff", "ColorLight") + and datapoint.get("type") == "switch" + ] if entities: async_add_entities(entities, update_before_add=True) @@ -48,42 +75,64 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry, asyn class JungHomeLight(CoordinatorEntity, LightEntity): """Representation of a Jung Home light.""" - def __init__(self, coordinator, device, datapoint): + def __init__( + self, + coordinator: JungHomeDataUpdateCoordinator, + device: Mapping[str, Any], + datapoint: Mapping[str, Any], + ) -> None: """Initialize the light.""" super().__init__(coordinator) self._device = device self._datapoint = datapoint # Find related datapoints (brightness / color_temperature) for ColorLight self._brightness_datapoint = next( - (dp for dp in device.get('datapoints', []) if dp.get('type') == 'brightness'), + ( + dp + for dp in device.get("datapoints", []) + if dp.get("type") == "brightness" + ), None, ) self._color_temp_datapoint = next( - (dp for dp in device.get('datapoints', []) if dp.get('type') == 'color_temperature'), + ( + dp + for dp in device.get("datapoints", []) + if dp.get("type") == "color_temperature" + ), None, ) self._brightness_datapoint_id = ( - self._brightness_datapoint.get('id') if self._brightness_datapoint else None + self._brightness_datapoint.get("id") if self._brightness_datapoint else None ) self._color_temp_datapoint_id = ( - self._color_temp_datapoint.get('id') if self._color_temp_datapoint else None + self._color_temp_datapoint.get("id") if self._color_temp_datapoint else None ) # Device brightness scale is 0-100 (device) — Home Assistant uses 0-255 # Track last local write to debounce weird rapid WS echoes - self._last_written_brightness_raw = None + self._last_written_brightness_raw: int | None = None self._last_written_brightness_ts = 0.0 # Track last local write for color temperature (Kelvin) - self._last_written_color_temp_raw = None + self._last_written_color_temp_raw: int | None = None self._last_written_color_temp_ts = 0.0 self._name = device.get("label", "Jung Light") - self._unique_id = f"{device.get('id')}_{datapoint.get('id')}" # Use device ID and datapoint ID + self._unique_id = f"{device.get('id')}_{datapoint.get('id')}" self._is_on = self._get_state_from_datapoint(datapoint) - self.entity_id = f"light.{self._unique_id}" # Set the entity ID - - if device['type'] == 'ColorLight': - # Read brightness and color_temp from their specific datapoints (if present) - self._brightness = self._get_brightness_from_datapoint(self._brightness_datapoint) - self._color_temp = self._get_color_temp_from_datapoint(self._color_temp_datapoint) + self.entity_id = f"light.{self._unique_id}" + + if device.get("type") == "ColorLight": + # Read brightness and color_temp from their specific datapoints + # (if present) + self._brightness = ( + self._get_brightness_from_datapoint(self._brightness_datapoint) + if self._brightness_datapoint + else 0 + ) + self._color_temp = ( + self._get_color_temp_from_datapoint(self._color_temp_datapoint) + if self._color_temp_datapoint + else None + ) self._attr_min_color_temp_kelvin = DEFAULT_MIN_KELVIN self._attr_max_color_temp_kelvin = DEFAULT_MAX_KELVIN else: @@ -91,27 +140,27 @@ def __init__(self, coordinator, device, datapoint): self._color_temp = None @property - def name(self): + def name(self) -> str: """Return the name of the light.""" return self._name @property - def unique_id(self): + def unique_id(self) -> str: """Return a unique ID for the light.""" return self._unique_id @property - def is_on(self): + def is_on(self) -> bool: """Return the state of the light.""" return self._is_on @property - def brightness(self): + def brightness(self) -> int | None: """Return the brightness of the light.""" return self._brightness @property - def color_temp(self): + def color_temp(self) -> int | None: """Return the color temperature of the light.""" # Home Assistant expects color_temp in mireds; device reports Kelvin if self._color_temp is None: @@ -129,14 +178,14 @@ def max_mireds(self) -> int: return kelvin_to_mired(DEFAULT_MIN_KELVIN) @property - def supported_color_modes(self): + def supported_color_modes(self) -> set[ColorMode]: """Return the supported color modes (only the current mode, per HA 2025.3+).""" return {self.color_mode} @property - def color_mode(self): + def color_mode(self) -> ColorMode: """Return the currently active color mode.""" - if self._device['type'] == 'ColorLight': + if self._device.get("type") == "ColorLight": # If color_temp is set, prefer COLOR_TEMP, else BRIGHTNESS if self._color_temp is not None: return ColorMode.COLOR_TEMP @@ -144,7 +193,7 @@ def color_mode(self): return ColorMode.ONOFF @property - def device_info(self): + def device_info(self) -> dict[str, Any]: """Return device information about this light.""" return { "identifiers": {(DOMAIN, self._device["id"])}, # Link to the device @@ -159,99 +208,123 @@ def _handle_coordinator_update(self) -> None: """Handle updated data from the coordinator.""" _LOGGER.debug("Handling coordinator update for light %s", self._name) device = next( - (d for d in self.coordinator.data if d["id"] == self._device["id"]), None + (d for d in self.coordinator.data if d["id"] == self._device["id"]), + None, ) if device: datapoint = next( - (dp for dp in device.get('datapoints', []) if dp["id"] == self._datapoint["id"]), None + ( + dp + for dp in device.get("datapoints", []) + if dp["id"] == self._datapoint["id"] + ), + None, ) if datapoint: self._is_on = self._get_state_from_datapoint(datapoint) - if self._device['type'] == 'ColorLight': - # Update brightness/color_temp from their respective datapoints (if available) + if self._device.get("type") == "ColorLight": if self._brightness_datapoint_id: - brightness_dp = next( - (dp for dp in device.get('datapoints', []) if dp.get('id') == self._brightness_datapoint_id), - None, - ) - # Read raw brightness value first - raw_brightness = None - if brightness_dp: - for v in brightness_dp.get('values', []): - if v.get('key') == 'brightness': - try: - raw_brightness = int(v.get('value')) - except (TypeError, ValueError): - raw_brightness = None - break - # If we recently wrote a brightness, debounce transient device - # echoes for a short window unless the echo matches our write. - now_ts = time.monotonic() - debounce_window = 3.0 - if ( - raw_brightness is not None - and self._last_written_brightness_raw is not None - and (now_ts - self._last_written_brightness_ts) < debounce_window - ): - if raw_brightness != self._last_written_brightness_raw: - _LOGGER.debug( - "Ignoring transient brightness echo %s for %s (recent write %s)", - raw_brightness, - self._name, - self._last_written_brightness_raw, - ) - # keep local self._brightness until confirmed - else: - # device echoed the same value we wrote — accept and clear tracking - self._brightness = self._get_brightness_from_datapoint(brightness_dp) - self._last_written_brightness_raw = None - self._last_written_brightness_ts = 0.0 - else: - self._brightness = self._get_brightness_from_datapoint(brightness_dp) + self._process_brightness_datapoint(device) if self._color_temp_datapoint_id: - color_dp = next( - (dp for dp in device.get('datapoints', []) if dp.get('id') == self._color_temp_datapoint_id), - None, - ) - # read raw Kelvin value - raw_kelvin = None - if color_dp: - for v in color_dp.get('values', []): - if v.get('key') == 'color_temperature': - try: - raw_kelvin = int(v.get('value')) - except (TypeError, ValueError): - raw_kelvin = None - break - # debounce transient color temp echoes similar to brightness - now_ts = time.monotonic() - debounce_window = 3.0 - if ( - raw_kelvin is not None - and self._last_written_color_temp_raw is not None - and (now_ts - self._last_written_color_temp_ts) < debounce_window - and raw_kelvin != self._last_written_color_temp_raw - ): - _LOGGER.debug( - "Ignoring transient color_temp echo %sK for %s (recent write %sK)", - raw_kelvin, - self._name, - self._last_written_color_temp_raw, - ) - # keep local value until confirmed - else: - self._color_temp = self._get_color_temp_from_datapoint(color_dp) + self._process_color_temp_datapoint(device) _LOGGER.debug("Updated state for light %s: %s", self._name, self._is_on) self.async_write_ha_state() - async def async_turn_on(self, **kwargs): + def _process_brightness_datapoint(self, device: Mapping[str, Any]) -> None: + """Process brightness datapoint updates and debounce echoes.""" + brightness_dp = next( + ( + dp + for dp in device.get("datapoints", []) + if dp.get("id") == self._brightness_datapoint_id + ), + None, + ) + raw_brightness = None + if brightness_dp: + for v in brightness_dp.get("values", []): + if v.get("key") == "brightness": + try: + raw_brightness = int(v.get("value")) + except (TypeError, ValueError): + raw_brightness = None + break + + now_ts = time.monotonic() + debounce_window = 3.0 + recent_write = ( + self._last_written_brightness_raw is not None + and (now_ts - self._last_written_brightness_ts) < debounce_window + ) + + if raw_brightness is not None and recent_write: + if raw_brightness != self._last_written_brightness_raw: + _LOGGER.debug( + "Ignoring transient brightness echo %s for %s (recent write %s)", + raw_brightness, + self._name, + self._last_written_brightness_raw, + ) + # keep local self._brightness until confirmed + else: + # device echoed the same value we wrote — accept and clear tracking + self._brightness = self._get_brightness_from_datapoint(brightness_dp) + self._last_written_brightness_raw = None + self._last_written_brightness_ts = 0.0 + else: + self._brightness = self._get_brightness_from_datapoint(brightness_dp) + + def _process_color_temp_datapoint(self, device: Mapping[str, Any]) -> None: + """Process color temperature datapoint updates and debounce echoes.""" + color_dp = next( + ( + dp + for dp in device.get("datapoints", []) + if dp.get("id") == self._color_temp_datapoint_id + ), + None, + ) + raw_kelvin = None + if color_dp: + for v in color_dp.get("values", []): + if v.get("key") == "color_temperature": + try: + raw_kelvin = int(v.get("value")) + except (TypeError, ValueError): + raw_kelvin = None + break + + now_ts = time.monotonic() + debounce_window = 3.0 + recent_write = ( + self._last_written_color_temp_raw is not None + and (now_ts - self._last_written_color_temp_ts) < debounce_window + ) + + if ( + raw_kelvin is not None + and recent_write + and raw_kelvin != self._last_written_color_temp_raw + ): + _LOGGER.debug( + "Ignoring transient color_temp echo %sK for %s (recent write %sK)", + raw_kelvin, + self._name, + self._last_written_color_temp_raw, + ) + # keep local value until confirmed + else: + self._color_temp = self._get_color_temp_from_datapoint(color_dp) + + async def async_turn_on(self, **kwargs: Any) -> None: """Turn the light on.""" + _ = kwargs _LOGGER.debug("Turning on light %s", self._name) # Turn on first, then apply brightness/color temperature to avoid # device-side overrides (some devices reset brightness on power-on). await self.coordinator.turn_on_light(self._datapoint["id"]) self._is_on = True - if self._device['type'] == 'ColorLight': + if self._device.get("type") == "ColorLight": if "brightness" in kwargs: brightness = kwargs["brightness"] await self._set_brightness(brightness) @@ -260,38 +333,39 @@ async def async_turn_on(self, **kwargs): await self._set_color_temp(color_temp) self.async_write_ha_state() - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Turn the light off.""" + _ = kwargs _LOGGER.debug("Turning off light %s", self._name) await self.coordinator.turn_off_light(self._datapoint["id"]) self._is_on = False self.async_write_ha_state() @property - def should_poll(self): + def should_poll(self) -> bool: """No polling needed for this entity.""" return False @property - def available(self): + def available(self) -> bool: """Return if the device is available.""" return self.coordinator.last_update_success - def _get_state_from_datapoint(self, datapoint): + def _get_state_from_datapoint(self, datapoint: Mapping[str, Any]) -> bool: """Extract the state of the light from its datapoint.""" - for value in datapoint.get('values', []): - if value['key'] == 'switch': - return value['value'] == '1' + for value in datapoint.get("values", []): + if value["key"] == "switch": + return value["value"] == "1" return False - def _get_brightness_from_datapoint(self, datapoint): + def _get_brightness_from_datapoint(self, datapoint: Mapping[str, Any]) -> int: """Extract the brightness of the light from its datapoint.""" if not datapoint: return 0 - for value in datapoint.get('values', []): - if value['key'] == 'brightness': + for value in datapoint.get("values", []): + if value["key"] == "brightness": try: - raw = int(value['value']) + raw = int(value["value"]) except (TypeError, ValueError): raw = 0 # Device reports 0-100; convert linearly to HA 0-255 @@ -302,30 +376,30 @@ def _raw_to_ha_brightness(self, raw: int) -> int: """Convert device raw brightness (0-100) to Home Assistant 0-255 scale.""" try: return round(raw * 255 / 100) - except Exception: + except (TypeError, ValueError): return 0 def _ha_to_raw_brightness(self, ha_brightness: int) -> int: """Convert Home Assistant 0-255 brightness to device raw scale (0-100).""" try: return round(int(ha_brightness) * 100 / 255) - except Exception: - return round(int(ha_brightness) * 100 / 255) + except (TypeError, ValueError): + return 0 - def _get_color_temp_from_datapoint(self, datapoint): + def _get_color_temp_from_datapoint(self, datapoint: Mapping[str, Any]) -> int: """Extract the color temperature of the light from its datapoint.""" if not datapoint: return 3000 - for value in datapoint.get('values', []): - if value['key'] == 'color_temperature': + for value in datapoint.get("values", []): + if value["key"] == "color_temperature": try: # Device reports Kelvin; store Kelvin - return int(value['value']) + return int(value["value"]) except (TypeError, ValueError): return 3000 return 3000 - async def _set_brightness(self, brightness): + async def _set_brightness(self, brightness: int) -> None: """Set the brightness of the light.""" _LOGGER.debug("Setting brightness for light %s to %s", self._name, brightness) if not self._brightness_datapoint_id: @@ -334,27 +408,44 @@ async def _set_brightness(self, brightness): # Convert Home Assistant 0-255 brightness to device raw scale (0-100 or 0-255) ha_brightness = int(brightness) raw_value = self._ha_to_raw_brightness(ha_brightness) - _LOGGER.debug("Converted HA brightness %s -> raw %s for %s", ha_brightness, raw_value, self._name) + _LOGGER.debug( + "Converted HA brightness %s -> raw %s for %s", + ha_brightness, + raw_value, + self._name, + ) await self.coordinator.set_brightness(self._brightness_datapoint_id, raw_value) # Record last write to debounce device echoes try: self._last_written_brightness_raw = int(raw_value) - except Exception: + except (TypeError, ValueError): self._last_written_brightness_raw = raw_value self._last_written_brightness_ts = time.monotonic() self._brightness = brightness self.async_write_ha_state() - async def _set_color_temp(self, color_temp): + async def _set_color_temp(self, color_temp: int) -> None: """Set the color temperature of the light.""" - _LOGGER.debug("Setting color temperature for light %s to %s", self._name, color_temp) + _LOGGER.debug( + "Setting color temperature for light %s to %s", + self._name, + color_temp, + ) if not self._color_temp_datapoint_id: - _LOGGER.warning("No color_temperature datapoint id for light %s", self._name) + _LOGGER.warning( + "No color_temperature datapoint id for light %s", + self._name, + ) return # Home Assistant supplies mireds; convert to Kelvin for device kelvin = mired_to_kelvin(int(color_temp)) - _LOGGER.debug("Converted HA color_temp %s mired -> %s K for %s", color_temp, kelvin, self._name) + _LOGGER.debug( + "Converted HA color_temp %s mired -> %s K for %s", + color_temp, + kelvin, + self._name, + ) await self.coordinator.set_color_temp(self._color_temp_datapoint_id, kelvin) # store Kelvin locally self._color_temp = kelvin - self.async_write_ha_state() \ No newline at end of file + self.async_write_ha_state() diff --git a/custom_components/junghome/manifest.json b/custom_components/junghome/manifest.json index e62dddd..6344f1d 100644 --- a/custom_components/junghome/manifest.json +++ b/custom_components/junghome/manifest.json @@ -1,6 +1,8 @@ { "domain": "junghome", "name": "Jung Home", + "documentation": "https://github.com/ernetas/junghome/blob/main/README.md", + "issue_tracker": "https://github.com/ernetas/junghome/issues", "config_flow": true, "version": "1.0.0", "requirements": [], diff --git a/custom_components/junghome/sensor.py b/custom_components/junghome/sensor.py index 61210e9..211ee86 100644 --- a/custom_components/junghome/sensor.py +++ b/custom_components/junghome/sensor.py @@ -1,13 +1,22 @@ +"""Sensor entities for the Junghome integration.""" + import logging +from collections.abc import Callable, Mapping +from typing import Any + from homeassistant.components.sensor import SensorEntity -from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.update_coordinator import CoordinatorEntity + from .const import DOMAIN _LOGGER = logging.getLogger(__name__) -async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities): + +async def async_setup_entry( + hass: HomeAssistant, entry: ConfigEntry, async_add_entities: Callable +) -> None: """Set up Jung Home sensors from a config entry.""" coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"] @@ -16,55 +25,86 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e devices = coordinator.data # Create sensor entities for each device - entities = [] + entities: list[SensorEntity] = [] for device in devices: - if device['type'] == 'Socket': # Add devices with type "Socket" - for datapoint in device.get('datapoints', []): - if datapoint['type'] == 'quantity': - label = next((value['value'].strip() for value in datapoint['values'] if value['key'] == 'quantity_label'), None) - unit = next((value['value'] for value in datapoint['values'] if value['key'] == 'quantity_unit'), None) + if device["type"] == "Socket": # Add devices with type "Socket" + for datapoint in device.get("datapoints", []): + if datapoint.get("type") == "quantity": + label = next( + ( + value["value"].strip() + for value in datapoint.get("values", []) + if value["key"] == "quantity_label" + ), + None, + ) + unit = next( + ( + value["value"] + for value in datapoint.get("values", []) + if value["key"] == "quantity_unit" + ), + None, + ) if label and unit: - entities.append(JungHomeQuantity(coordinator, device, datapoint, label, unit)) + entities.append( + JungHomeQuantity( + coordinator, + device, + datapoint, + label, + unit, + ) + ) if entities: async_add_entities(entities, update_before_add=True) + class JungHomeQuantity(CoordinatorEntity, SensorEntity): """Representation of a Jung Home quantity.""" - def __init__(self, coordinator, device, datapoint, label, unit): + def __init__( + self, + coordinator: Any, + device: dict[str, Any], + datapoint: Mapping[str, Any], + label: str, + unit: str, + ) -> None: """Initialize the quantity.""" super().__init__(coordinator) self._device = device self._datapoint = datapoint self._name = f"{device.get('label', 'Jung Device')} {label}" - self._unique_id = f"{device.get('id')}_{datapoint.get('id')}_{label.replace(' ', '_').lower()}" + normalized_label = label.replace(" ", "_").lower() + self._unique_id = f"{device.get('id')}_{datapoint.get('id')}_{normalized_label}" self._unit = unit self._value = self._get_value_from_datapoint(datapoint) self.entity_id = f"sensor.{self._unique_id}" # Set the entity ID @property - def name(self): + def name(self) -> str: """Return the name of the quantity.""" return self._name @property - def unique_id(self): + def unique_id(self) -> str: """Return a unique ID for the quantity.""" return self._unique_id @property - def state(self): + def state(self) -> Any: """Return the state of the quantity.""" return self._value @property - def unit_of_measurement(self): + def unit_of_measurement(self) -> str: """Return the unit of measurement of the quantity.""" return self._unit @property - def device_info(self): + def device_info(self) -> dict[str, Any]: """Return device information about this quantity.""" return { "identifiers": {(DOMAIN, self._device["id"])}, # Link to the device @@ -79,20 +119,30 @@ def _handle_coordinator_update(self) -> None: """Handle updated data from the coordinator.""" _LOGGER.debug("Handling coordinator update for quantity %s", self._name) device = next( - (d for d in self.coordinator.data if d["id"] == self._device["id"]), None + (d for d in self.coordinator.data if d["id"] == self._device["id"]), + None, ) if device: datapoint = next( - (dp for dp in device.get('datapoints', []) if dp["id"] == self._datapoint["id"]), None + ( + dp + for dp in device.get("datapoints", []) + if dp["id"] == self._datapoint["id"] + ), + None, ) if datapoint: self._value = self._get_value_from_datapoint(datapoint) - _LOGGER.debug("Updated state for quantity %s: %s", self._name, self._value) + _LOGGER.debug( + "Updated state for quantity %s: %s", + self._name, + self._value, + ) self.async_write_ha_state() - def _get_value_from_datapoint(self, datapoint): + def _get_value_from_datapoint(self, datapoint: Mapping[str, Any]) -> Any: """Extract the value of the quantity from its datapoint.""" - for value in datapoint.get('values', []): - if value['key'] == 'quantity': - return value['value'] - return None \ No newline at end of file + for value in datapoint.get("values", []): + if value["key"] == "quantity": + return value["value"] + return None diff --git a/custom_components/junghome/switch.py b/custom_components/junghome/switch.py index ff1711c..d6511c2 100644 --- a/custom_components/junghome/switch.py +++ b/custom_components/junghome/switch.py @@ -1,13 +1,22 @@ +"""Switch entities for the Junghome integration.""" + import logging +from collections.abc import Callable, Mapping +from typing import Any + from homeassistant.components.switch import SwitchEntity -from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback -from homeassistant.helpers.typing import ConfigType +from homeassistant.helpers.update_coordinator import CoordinatorEntity + from .const import DOMAIN _LOGGER = logging.getLogger(__name__) -async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities): + + +async def async_setup_entry( + hass: HomeAssistant, entry: ConfigEntry, async_add_entities: Callable +) -> None: """Set up Jung Home switches from a config entry.""" coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"] @@ -18,53 +27,60 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e # Create switch entities for each device entities = [] for device in devices: - if device['type'] == 'Socket': # Add devices with type "Socket" - for datapoint in device.get('datapoints', []): - if datapoint['type'] == 'switch': - entities.append(JungHomeSocket(coordinator, device, datapoint)) - elif device.get('type') == 'RockerSwitch': - for datapoint in device.get('datapoints', []): - if datapoint.get('type') == 'status_led': - entities.append(JungHomeSwitch(coordinator, device, datapoint)) + if device["type"] == "Socket": + entities.extend( + JungHomeSocket(coordinator, device, datapoint) + for datapoint in device.get("datapoints", []) + if datapoint.get("type") == "switch" + ) + elif device.get("type") == "RockerSwitch": + entities.extend( + JungHomeSwitch(coordinator, device, datapoint) + for datapoint in device.get("datapoints", []) + if datapoint.get("type") == "status_led" + ) if entities: async_add_entities(entities, update_before_add=True) + class JungHomeSocket(CoordinatorEntity, SwitchEntity): """Representation of a Jung Home socket.""" - def __init__(self, coordinator, device, datapoint): + def __init__( + self, coordinator: Any, device: dict[str, Any], datapoint: Mapping[str, Any] + ) -> None: """Initialize the socket.""" super().__init__(coordinator) self._device = device self._datapoint = datapoint self._name = device.get("label", "Jung Socket") - self._unique_id = f"{device.get('id')}_{datapoint.get('id')}" # Use device ID and datapoint ID + self._unique_id = f"{device.get('id')}_{datapoint.get('id')}" self._is_on = self._get_state_from_datapoint(datapoint) self.entity_id = f"switch.{self._unique_id}" # Set the entity ID @property - def name(self): + def name(self) -> str: """Return the name of the socket.""" return self._name @property - def unique_id(self): + def unique_id(self) -> str: """Return a unique ID for the socket.""" return self._unique_id @property - def is_on(self): + def is_on(self) -> bool: """Return the state of the socket.""" return self._is_on @property - def device_class(self): + def device_class(self) -> str: """Return the device class of the socket.""" return "outlet" @property - def device_info(self): + def device_info(self) -> dict[str, Any]: """Return device information about this socket.""" return { "identifiers": {(DOMAIN, self._device["id"])}, # Link to the device @@ -79,75 +95,102 @@ def _handle_coordinator_update(self) -> None: """Handle updated data from the coordinator.""" _LOGGER.debug("Handling coordinator update for socket %s", self._name) device = next( - (d for d in self.coordinator.data if d["id"] == self._device["id"]), None + (d for d in self.coordinator.data if d["id"] == self._device["id"]), + None, ) if device: datapoint = next( - (dp for dp in device.get('datapoints', []) if dp["id"] == self._datapoint["id"]), None + ( + dp + for dp in device.get("datapoints", []) + if dp["id"] == self._datapoint["id"] + ), + None, ) if datapoint: self._is_on = self._get_state_from_datapoint(datapoint) - _LOGGER.debug("Updated state for socket %s: %s", self._name, self._is_on) + _LOGGER.debug( + "Updated state for socket %s: %s", + self._name, + self._is_on, + ) self.async_write_ha_state() - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Turn the socket on.""" + _ = kwargs _LOGGER.debug("Turning on socket %s", self._name) await self.coordinator.turn_on_switch(self._datapoint["id"]) self._is_on = True self.async_write_ha_state() - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Turn the socket off.""" + _ = kwargs _LOGGER.debug("Turning off socket %s", self._name) await self.coordinator.turn_off_switch(self._datapoint["id"]) self._is_on = False self.async_write_ha_state() @property - def should_poll(self): + def should_poll(self) -> bool: """No polling needed for this entity.""" return False @property - def available(self): + def available(self) -> bool: """Return if the device is available.""" return self.coordinator.last_update_success - async def async_update(self): + async def async_update(self) -> None: """Update the socket's state.""" await self.coordinator.async_request_refresh() device = next( - (d for d in self.coordinator.data if d["id"] == self._device["id"]), None + (d for d in self.coordinator.data if d["id"] == self._device["id"]), + None, ) if device: datapoint = next( - (dp for dp in device.get('datapoints', []) if dp["id"] == self._datapoint["id"]), None + ( + dp + for dp in device.get("datapoints", []) + if dp["id"] == self._datapoint["id"] + ), + None, ) if datapoint: self._is_on = self._get_state_from_datapoint(datapoint) self.async_write_ha_state() - def _get_state_from_datapoint(self, datapoint): + def _get_state_from_datapoint(self, datapoint: Mapping[str, Any]) -> bool: """Extract the state of the socket from its datapoint.""" - for value in datapoint.get('values', []): - if value['key'] == 'switch': - return value['value'] == '1' + for value in datapoint.get("values", []): + if value["key"] == "switch": + return value["value"] == "1" return False + class JungHomeSwitch(CoordinatorEntity, SwitchEntity): """Representation of a Jung Home status LED as a switch entity.""" - def __init__(self, coordinator, device, datapoint): + + def __init__( + self, coordinator: Any, device: dict[str, Any], datapoint: Mapping[str, Any] + ) -> None: + """Initialize the status LED switch entity.""" super().__init__(coordinator) self._device = device self._datapoint = datapoint - self._attr_name = f"{device.get('label', 'Jung Status LED')}_{datapoint.get('type', 'Unknown')}" + # Build a readable name for the entity + label = device.get("label", "Jung Status LED") + dp_type = datapoint.get("type", "Unknown") + self._attr_name = f"{label}_{dp_type}" self._attr_unique_id = f"{device.get('id')}_{datapoint.get('id')}_switch" self._attr_available = coordinator.last_update_success self._attr_is_on = self._get_state_from_datapoint(datapoint) @property - def device_info(self): + def device_info(self) -> dict[str, Any]: + """Return device information for the parent device.""" return { "identifiers": {(DOMAIN, self._device["id"])}, "name": self._device.get("label", "Jung Device"), @@ -156,34 +199,49 @@ def device_info(self): "sw_version": self._device.get("sw_version", "Unknown Version"), } - def _get_state_from_datapoint(self, datapoint): - for value in datapoint.get('values', []): - if value['key'] == 'status_led': - return value['value'] == '1' + def _get_state_from_datapoint(self, datapoint: Mapping[str, Any]) -> bool: + """Return True if the datapoint indicates the LED is enabled.""" + for value in datapoint.get("values", []): + if value["key"] == "status_led": + return value["value"] == "1" return False @property - def is_on(self): + def is_on(self) -> bool: + """Return if the status LED is on.""" return self._attr_is_on - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: + """Turn the status LED on.""" + _ = kwargs _LOGGER.debug("Turning on switch %s", self._attr_name) - await self.coordinator.set_status_led(self._datapoint["id"], True) + await self.coordinator.set_status_led(self._datapoint["id"], state=True) - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: + """Turn the status LED off.""" + _ = kwargs _LOGGER.debug("Turning off switch %s", self._attr_name) - await self.coordinator.set_status_led(self._datapoint["id"], False) + await self.coordinator.set_status_led(self._datapoint["id"], state=False) @callback def _handle_coordinator_update(self) -> None: _LOGGER.debug("Updating switch for %s", self._attr_name) - device = next((d for d in self.coordinator.data if d["id"] == self._device["id"]), None) + device = next( + (d for d in self.coordinator.data if d["id"] == self._device["id"]), None + ) if not device: return - datapoint = next((dp for dp in device.get("datapoints", []) if dp["id"] == self._datapoint["id"]), None) + datapoint = next( + ( + dp + for dp in device.get("datapoints", []) + if dp["id"] == self._datapoint["id"] + ), + None, + ) if not datapoint: return new_state = self._get_state_from_datapoint(datapoint) if new_state != self._attr_is_on: self._attr_is_on = new_state - self.async_write_ha_state() \ No newline at end of file + self.async_write_ha_state() From 45a0e435fd6aad9b4ccf1a76bcf1a5918c0af6b5 Mon Sep 17 00:00:00 2001 From: Ernestas Date: Tue, 30 Dec 2025 23:00:24 +0200 Subject: [PATCH 2/4] fix --- custom_components/junghome/manifest.json | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/custom_components/junghome/manifest.json b/custom_components/junghome/manifest.json index 6344f1d..8484586 100644 --- a/custom_components/junghome/manifest.json +++ b/custom_components/junghome/manifest.json @@ -7,5 +7,6 @@ "version": "1.0.0", "requirements": [], "dependencies": [], - "codeowners": [] + "codeowners": [], + "iot_class": "local_push" } From 0bd4ad04068ed5d3d5f5a404f05ef52aef98ec88 Mon Sep 17 00:00:00 2001 From: Ernestas Date: Tue, 30 Dec 2025 23:01:15 +0200 Subject: [PATCH 3/4] fix --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index ed27629..9ffa3ec 100644 --- a/README.md +++ b/README.md @@ -19,7 +19,7 @@ All communication is via WebSockets. I've managed to reliably automate: Any feedback is welcome, this is my first integration with HomeAssistant. # How -When adding the integration, you will need to know your Jung Home gateway address and get an API token from https:///api/junghome/swagger/ (using User Registration action). +When adding the integration, you will need to know your Jung Home gateway address and get an API token from `https:///api/junghome/swagger/` (using User Registration action). You will need to confirm the token in the Jung Home mobile app. From 0423abf330e1f7c64135f591bf99e781f7489658 Mon Sep 17 00:00:00 2001 From: Ernestas Date: Tue, 30 Dec 2025 23:03:29 +0200 Subject: [PATCH 4/4] fix --- custom_components/junghome/manifest.json | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/custom_components/junghome/manifest.json b/custom_components/junghome/manifest.json index 8484586..658a0d2 100644 --- a/custom_components/junghome/manifest.json +++ b/custom_components/junghome/manifest.json @@ -1,12 +1,12 @@ { - "domain": "junghome", - "name": "Jung Home", + "codeowners": [], + "config_flow": true, + "dependencies": [], "documentation": "https://github.com/ernetas/junghome/blob/main/README.md", + "domain": "junghome", + "iot_class": "local_push", "issue_tracker": "https://github.com/ernetas/junghome/issues", - "config_flow": true, - "version": "1.0.0", + "name": "Jung Home", "requirements": [], - "dependencies": [], - "codeowners": [], - "iot_class": "local_push" + "version": "1.0.0" }