add redgtech integration

.strict-typing

@@ -405,6 +405,7 @@ homeassistant.components.raspberry_pi.*
 homeassistant.components.rdw.*
 homeassistant.components.recollect_waste.*
 homeassistant.components.recorder.*
+homeassistant.components.redgtech.*
 homeassistant.components.remote.*
 homeassistant.components.renault.*
 homeassistant.components.reolink.*

homeassistant/components/redgtech/__init__.py

@@ -0,0 +1,63 @@
+import logging
+from homeassistant.core import HomeAssistant
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from .const import DOMAIN, API_URL
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+import aiohttp
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORMS: list[Platform] = [Platform.SWITCH, Platform.LIGHT]
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Redgtech from a config entry."""
+    _LOGGER.debug("Setting up Redgtech entry: %s", entry.entry_id)
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][entry.entry_id] = {
+        "config": entry.data,
+        "entities": []
+    }
+
+    access_token = entry.data.get("access_token")
+    if not access_token:
+        _LOGGER.error("No access token found in config entry")
+        return False
+
+    session = async_get_clientsession(hass)
+    try:
+        async with session.get(f'{API_URL}/home_assistant?access_token={access_token}', timeout=10) as response:
+            response.raise_for_status()
+            data = await response.json()
+            _LOGGER.debug("Received data from API: %s", data)
+
+            entities = [
+                {
+                    "id": item.get('endpointId', ''),
+                    "name": item.get("name", f"Entity {item.get('endpointId', '')}"),
+                    "state": "on" if item.get("value", False) else "off",
+                    "brightness": item.get("bright", 0),
+                    "type": 'light' if 'dim' in item.get('endpointId', '').lower() else 'switch'
+                }
+                for item in data.get("boards", [])
+            ]
+            hass.data[DOMAIN][entry.entry_id]["entities"] = entities
+
+        await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+        _LOGGER.debug("Successfully set up Redgtech entry: %s", entry.entry_id)
+        return True
+
+    except aiohttp.ClientResponseError as e:
+        _LOGGER.error("HTTP error while setting up Redgtech entry: %s - Status: %s", e.message, e.status)
+        return False
+    except aiohttp.ClientError as e:
+        _LOGGER.error("Client error while setting up Redgtech entry: %s", e)
+        return False
+    except Exception as e:
+        _LOGGER.exception("Unexpected error setting up Redgtech entry: %s", entry.entry_id)
+        return False
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    _LOGGER.debug("Unloading Redgtech entry: %s", entry.entry_id)
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/redgtech/config_flow.py

@@ -0,0 +1,55 @@
+from homeassistant import config_entries
+import voluptuous as vol
+import aiohttp
+import logging
+from .const import DOMAIN, API_URL
+
+_LOGGER = logging.getLogger(__name__)
+
+class RedgtechConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Config Flow for Redgtech integration."""
+
+    VERSION = 1
+
+    async def async_step_user(self, user_input=None):
+        """Handle the initial user step for login."""
+        errors = {}
+
+        if user_input is not None:
+            email = user_input.get("email")
+            password = user_input.get("password")
+
+            try:
+                async with aiohttp.ClientSession() as session:
+                    async with session.post(
+                        f'{API_URL}/home_assistant/login',
+                        json={'email': email, 'password': password}
+                    ) as response:
+                        if response.status == 200:
+                            data = await response.json()
+                            access_token = data.get("data", {}).get("access_token")
+                            if access_token:
+                                _LOGGER.info("Login successful")
+
+                                return self.async_create_entry(
+                                    title="Redgtech",
+                                    data={"access_token": access_token}
+                                )
+                            else:
+                                _LOGGER.error("Login failed: No access token received")
+                                errors["base"] = "invalid_auth"
+                        else:
+                            _LOGGER.error("Login failed: Invalid credentials")
+                            errors["base"] = "invalid_auth"
+            except aiohttp.ClientError as e:
+                _LOGGER.error("Login failed: Cannot connect to server: %s", e)
+                errors["base"] = "cannot_connect"
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema({
+                vol.Required("email"): str,
+                vol.Required("password"): str,
+            }),
+            errors=errors
+        )

homeassistant/components/redgtech/const.py

@@ -0,0 +1,2 @@
+DOMAIN = "redgtech"
+API_URL = "https://redgtech-dev.com"

homeassistant/components/redgtech/light.py

@@ -0,0 +1,127 @@
+from homeassistant.components.light import LightEntity, ColorMode
+from homeassistant.const import STATE_ON, STATE_OFF, CONF_BRIGHTNESS
+from .const import API_URL
+import aiohttp
+import logging
+
+_LOGGER = logging.getLogger(__name__)
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up the light platform."""
+    access_token = config_entry.data.get("access_token")
+    if access_token:
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(f'{API_URL}/home_assistant?access_token={access_token}') as response:
+                    if response.status == 200:
+                        data = await response.json()
+                        entities = []
+
+                        for item in data.get("boards", []):
+                            endpoint_id = item.get('endpointId', '')
+                            if 'dim' in endpoint_id:
+                                entities.append(RedgtechLight(item, access_token))
+
+                        async_add_entities(entities)
+                    else:
+                        _LOGGER.error("Error fetching data from API: %s", response.status)
+        except aiohttp.ClientError as e:
+            _LOGGER.error("Error connecting to API: %s", e)
+    else:
+        _LOGGER.error("No access token available")
+
+
+class RedgtechLight(LightEntity):
+    """Representation of a dimmable light."""
+
+    def __init__(self, data, token):
+        self._state = STATE_ON if data.get("value", False) else STATE_OFF
+        self._brightness = self._convert_brightness(data.get("bright", 0))
+        self._previous_brightness = self._brightness
+        self._name = data.get("friendlyName")
+        self._endpoint_id = data.get("endpointId")
+        self._description = data.get("description")
+        self._manufacturer = data.get("manufacturerName")
+        self._token = token
+        self._supported_color_modes = {ColorMode.BRIGHTNESS}
+        self._color_mode = ColorMode.BRIGHTNESS
+
+    @property
+    def name(self):
+        """Return the name of the light."""
+        return self._name
+
+    @property
+    def is_on(self):
+        """Return true if the light is on."""
+        return self._state == STATE_ON
+
+    @property
+    def brightness(self):
+        """Return the brightness of the light."""
+        return self._brightness
+
+    @property
+    def supported_color_modes(self):
+        """Return supported color modes."""
+        return self._supported_color_modes
+
+    @property
+    def color_mode(self):
+        """Return the color mode of the light."""
+        return self._color_mode
+
+    async def async_turn_on(self, **kwargs):
+        """Turn the light on with optional brightness."""
+        brightness = kwargs.get(CONF_BRIGHTNESS, self._previous_brightness)
+        await self._set_state(STATE_ON, brightness)
+
+    async def async_turn_off(self, **kwargs):
+        """Turn the light off."""
+        self._previous_brightness = self._brightness
+        await self._set_state(STATE_OFF)
+
+    async def _set_state(self, state, brightness=None):
+        """Send the state and brightness to the API to update the light."""
+        id_part, after_id = self._endpoint_id.split("-", 1)
+        number_channel = after_id[-1]
+        type_channel = ''.join(char for char in after_id if char.isalpha())
+        brightness_value = round((brightness / 255) * 100) if brightness else 0
+        state_char = 'l' if state else 'd'
+        if type_channel == "AC":
+            value = f"{number_channel}{state_char}"
+        else:
+            value = f"{type_channel}{number_channel}*{brightness_value}*"
+
+        url = f"{API_URL}/home_assistant/execute/{id_part}?cod=?{value}"
+        headers = {"Authorization": f"{self._token}"}
+        payload = {"state": state}
+
+        async with aiohttp.ClientSession() as session:
+            async with session.get(url, headers=headers, json=payload) as response:
+                if response.status == 200:
+                    self._state = state
+                    if state == STATE_ON:
+                        self._brightness = brightness or 255
+                    else:
+                        self._brightness = 0
+                    self.async_write_ha_state()
+                else:
+                    _LOGGER.error("Failed to set state for %s, status code: %s", self._name, response.status)
+
+    @property
+    def extra_state_attributes(self):
+        """Return the state attributes."""
+        return {
+            "endpoint_id": self._endpoint_id,
+            "description": self._description,
+            "manufacturer": self._manufacturer,
+        }
+
+    def _convert_brightness(self, bright_value):
+        """Convert brightness value from 0-100 to 0-255."""
+        try:
+            return int((int(bright_value) / 100) * 255)
+        except (ValueError, TypeError):
+            _LOGGER.error("Invalid brightness value: %s", bright_value)
+            return 0

homeassistant/components/redgtech/manifest.json

@@ -0,0 +1,13 @@
+{
+    "domain": "redgtech",
+    "name": "Redgtech",
+    "version": "1.0.0",
+    "codeowners": [],
+    "documentation": "https://www.home-assistant.io/integrations/redgtech",
+    "iot_class": "cloud_polling",
+    "logo": "/brands/redgtech/logo.png",
+    "integration_type": "service",
+    "config_flow": true,
+    "quality_scale": "bronze",
+    "requirements": []
+}

homeassistant/components/redgtech/quality_scale.yaml

@@ -0,0 +1,90 @@
+rules:
+  # Bronze
+  action-setup:
+    status: exempt
+    comment: only entity actions
+  appropriate-polling:
+    status: exempt
+    comment: the integration does not poll
+  brands: done
+  common-modules:
+    status: exempt
+    comment: the integration currently implements only one platform and has no coordinator
+  config-flow-test-coverage: done
+  config-flow: done
+  dependency-transparency: done
+  docs-actions: done
+  docs-high-level-description: done
+  docs-installation-instructions: done
+  docs-removal-instructions: done
+  entity-event-setup:
+    status: exempt
+    comment: the integration does not subscribe to events
+  entity-unique-id: done
+  has-entity-name: done
+  runtime-data: done
+  test-before-configure:
+    status: done
+    comment: tested by publishing a success message to the topic
+  test-before-setup:
+    status: exempt
+    comment: testing would require to trigger a notification
+  unique-config-entry: done
+
+  # Silver
+  action-exceptions: done
+  config-entry-unloading: done
+  docs-configuration-parameters:
+    status: exempt
+    comment: the integration has no options
+  docs-installation-parameters: done
+  entity-unavailable:
+    status: exempt
+    comment: the integration only implements a stateless notify entity.
+  integration-owner: done
+  log-when-unavailable:
+    status: exempt
+    comment: the integration only integrates state-less entities
+  parallel-updates: done
+  reauthentication-flow:
+    status: exempt
+    comment: the integration currently does not implement authenticated requests
+  test-coverage: done
+
+  # Gold
+  devices: todo
+  diagnostics: todo
+  discovery-update-info: todo
+  discovery: todo
+  docs-data-update: todo
+  docs-examples: todo
+  docs-known-limitations: todo
+  docs-supported-devices: todo
+  docs-supported-functions: todo
+  docs-troubleshooting: todo
+  docs-use-cases: todo
+  dynamic-devices: todo
+  entity-category: done
+  entity-device-class:
+    status: exempt
+    comment: no suitable device class for the notify entity
+  entity-disabled-by-default:
+    status: exempt
+    comment: only one entity
+  entity-translations:
+    status: exempt
+    comment: the notify entity uses the topic as name, no translation required
+  exception-translations: done
+  icon-translations: done
+  reconfiguration-flow: todo
+  repair-issues:
+    status: exempt
+    comment: the integration has no repeairs
+  stale-devices:
+    status: exempt
+    comment: only one device per entry, is deleted with the entry.
+
+  # Platinum
+  async-dependency: done
+  inject-websession: done
+  strict-typing: done

homeassistant/components/redgtech/services.yaml

@@ -0,0 +1,12 @@
+login:
+  description: "Log in to the Redgtech service"
+  fields:
+    email:
+      description: "The email address for the Redgtech account"
+      example: "[email protected]"
+    password:
+      description: "The password for the Redgtech account"
+      example: "your_password"
+
+logout:
+  description: "Log out from the Redgtech service"