Yes, I replaced the HA Velux integration with the code in the PR. It works perfectly.
It’s really a pitty, that such PRs don’t get it into the official repo just because of being ignored for too long time.
In the directory /config/custom_components/velux
add these files, which will replace the core integration:
__init__.py
"""Support for VELUX KLF 200 devices."""
import logging
from pyvlx import OpeningDevice, Window, PyVLX, PyVLXException
import voluptuous as vol
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import ConfigType
DOMAIN = "velux"
DATA_VELUX = "data_velux"
PLATFORMS = [Platform.COVER, Platform.LIGHT, Platform.SCENE]
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.Schema(
{vol.Required(CONF_HOST): cv.string, vol.Required(CONF_PASSWORD): cv.string}
)
},
extra=vol.ALLOW_EXTRA,
)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the velux component."""
try:
hass.data[DATA_VELUX] = VeluxModule(hass, config[DOMAIN])
hass.data[DATA_VELUX].setup()
await hass.data[DATA_VELUX].async_start()
except PyVLXException as ex:
_LOGGER.exception("Can't connect to velux interface: %s", ex)
return False
for platform in PLATFORMS:
hass.async_create_task(
discovery.async_load_platform(hass, platform, DOMAIN, {}, config)
)
return True
class VeluxModule:
"""Abstraction for velux component."""
def __init__(self, hass, domain_config):
"""Initialize for velux component."""
self.pyvlx = None
self._hass = hass
self._domain_config = domain_config
def setup(self):
"""Velux component setup."""
async def on_hass_stop(event):
"""Close connection when hass stops."""
_LOGGER.debug("Velux interface terminated")
await self.pyvlx.disconnect()
async def async_reboot_gateway(service_call: ServiceCall) -> None:
await self.pyvlx.reboot_gateway()
self._hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, on_hass_stop)
host = self._domain_config.get(CONF_HOST)
password = self._domain_config.get(CONF_PASSWORD)
self.pyvlx = PyVLX(host=host, password=password)
self._hass.services.async_register(
DOMAIN, "reboot_gateway", async_reboot_gateway
)
async def async_start(self):
"""Start velux component."""
_LOGGER.debug("Velux interface started")
await self.pyvlx.load_scenes()
await self.pyvlx.load_nodes()
class VeluxEntity(Entity):
"""Abstraction for al Velux entities."""
_attr_should_poll = False
def __init__(self, node: OpeningDevice) -> None:
"""Initialize the Velux device."""
self.node = node
self._attr_unique_id = node.serial_number
self._attr_name = node.name if node.name else f"#{node.node_id}"
async def async_init(self):
"""Initialize the entity async."""
@callback
def async_register_callbacks(self):
"""Register callbacks to update hass after device was changed."""
async def after_update_callback(device):
"""Call after device was updated."""
self.async_write_ha_state()
self.node.register_device_updated_cb(after_update_callback)
async def async_added_to_hass(self):
"""Store register state change callback."""
self.async_register_callbacks()
cover.py
"""Support for Velux covers."""
from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any, TYPE_CHECKING
from pyvlx import OpeningDevice, Position
from pyvlx.exception import PyVLXException
from pyvlx.opening_device import Awning, Blind, GarageDoor, Gate, RollerShutter, Window
from homeassistant.components.cover import (
ATTR_POSITION,
ATTR_TILT_POSITION,
CoverDeviceClass,
CoverEntity,
CoverEntityFeature,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from . import DATA_VELUX, VeluxEntity
PARALLEL_UPDATES = 1
if TYPE_CHECKING:
from pyvlx.node import Node
DEFAULT_SCAN_INTERVAL = timedelta(minutes=2)
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up cover(s) for Velux platform."""
entities: list[VeluxWindow | VeluxCover] = []
for node in hass.data[DATA_VELUX].pyvlx.nodes:
if isinstance(node, OpeningDevice):
if isinstance(node, Window):
entities.append(VeluxWindow(hass, node))
else:
entities.append(VeluxCover(node))
for entity in entities:
await entity.async_init()
async_add_entities(entities)
class VeluxCover(VeluxEntity, CoverEntity):
"""Representation of a Velux cover."""
_is_blind = False
def __init__(self, node: OpeningDevice) -> None:
"""Initialize VeluxCover."""
super().__init__(node)
self._attr_device_class = CoverDeviceClass.WINDOW
if isinstance(node, Awning):
self._attr_device_class = CoverDeviceClass.AWNING
if isinstance(node, Blind):
self._attr_device_class = CoverDeviceClass.BLIND
self._is_blind = True
if isinstance(node, GarageDoor):
self._attr_device_class = CoverDeviceClass.GARAGE
if isinstance(node, Gate):
self._attr_device_class = CoverDeviceClass.GATE
if isinstance(node, RollerShutter):
self._attr_device_class = CoverDeviceClass.SHUTTER
if isinstance(node, Window):
self._attr_device_class = CoverDeviceClass.WINDOW
@property
def supported_features(self) -> CoverEntityFeature:
"""Flag supported features."""
supported_features = (
CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE
| CoverEntityFeature.SET_POSITION
| CoverEntityFeature.STOP
)
if self.current_cover_tilt_position is not None:
supported_features |= (
CoverEntityFeature.OPEN_TILT
| CoverEntityFeature.CLOSE_TILT
| CoverEntityFeature.SET_TILT_POSITION
| CoverEntityFeature.STOP_TILT
)
return supported_features
@property
def current_cover_position(self) -> int:
"""Return the current position of the cover."""
return 100 - self.node.position.position_percent
@property
def current_cover_tilt_position(self) -> int | None:
"""Return the current position of the cover."""
if self._is_blind:
return 100 - self.node.orientation.position_percent
return None
@property
def is_closed(self) -> bool:
"""Return if the cover is closed."""
return self.node.position.closed
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover."""
await self.node.close(wait_for_completion=False)
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
await self.node.open(wait_for_completion=False)
async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Move the cover to a specific position."""
position_percent = 100 - kwargs[ATTR_POSITION]
await self.node.set_position(
Position(position_percent=position_percent), wait_for_completion=False
)
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
await self.node.stop(wait_for_completion=False)
async def async_close_cover_tilt(self, **kwargs: Any) -> None:
"""Close cover tilt."""
await self.node.close_orientation(wait_for_completion=False)
async def async_open_cover_tilt(self, **kwargs: Any) -> None:
"""Open cover tilt."""
await self.node.open_orientation(wait_for_completion=False)
async def async_stop_cover_tilt(self, **kwargs: Any) -> None:
"""Stop cover tilt."""
await self.node.stop_orientation(wait_for_completion=False)
async def async_set_cover_tilt_position(self, **kwargs: Any) -> None:
"""Move cover tilt to a specific position."""
position_percent = 100 - kwargs[ATTR_TILT_POSITION]
orientation = Position(position_percent=position_percent)
await self.node.set_orientation(
orientation=orientation, wait_for_completion=False
)
class VeluxWindow(VeluxCover):
"""Representation of a Velux window."""
def __init__(self, hass: HomeAssistant, node: Node) -> None:
"""Initialize Velux window."""
super().__init__(node)
self._hass = hass
self._extra_attr_limitation_min: int | None = None
self._extra_attr_limitation_max: int | None = None
self.coordinator = DataUpdateCoordinator(
self._hass,
_LOGGER,
name=self.unique_id,
update_method=self.async_update_limitation,
update_interval=DEFAULT_SCAN_INTERVAL,
)
async def async_init(self):
"""Async initialize."""
return await self.coordinator.async_config_entry_first_refresh()
async def async_update_limitation(self):
"""Get the updated status of the cover (limitations only)."""
try:
limitation = await self.node.get_limitation()
self._extra_attr_limitation_min = limitation.min_value
self._extra_attr_limitation_max = limitation.max_value
except PyVLXException:
_LOGGER.error("Error fetch limitation data for cover %s", self.name)
@property
def extra_state_attributes(self) -> dict[str, int | None]:
"""Return the state attributes."""
return {
"limitation_min": self._extra_attr_limitation_min,
"limitation_max": self._extra_attr_limitation_max,
}
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
self.async_on_remove(
self.coordinator.async_add_listener(self._handle_coordinator_update)
)
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self.async_write_ha_state()
async def async_update(self) -> None:
"""Update the entity.
Only used by the generic entity update service.
"""
await self.coordinator.async_request_refresh()
light.py
"""Support for Velux lights."""
from __future__ import annotations
from typing import Any
from pyvlx import Intensity, LighteningDevice
from homeassistant.components.light import ATTR_BRIGHTNESS, ColorMode, LightEntity
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DATA_VELUX, VeluxEntity
PARALLEL_UPDATES = 1
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up light(s) for Velux platform."""
async_add_entities(
VeluxLight(node)
for node in hass.data[DATA_VELUX].pyvlx.nodes
if isinstance(node, LighteningDevice)
)
class VeluxLight(VeluxEntity, LightEntity):
"""Representation of a Velux light."""
_attr_supported_color_modes = {ColorMode.BRIGHTNESS}
_attr_color_mode = ColorMode.BRIGHTNESS
@property
def brightness(self):
"""Return the current brightness."""
return int((100 - self.node.intensity.intensity_percent) * 255 / 100)
@property
def is_on(self):
"""Return true if light is on."""
return not self.node.intensity.off and self.node.intensity.known
async def async_turn_on(self, **kwargs: Any) -> None:
"""Instruct the light to turn on."""
if ATTR_BRIGHTNESS in kwargs:
intensity_percent = int(100 - kwargs[ATTR_BRIGHTNESS] / 255 * 100)
await self.node.set_intensity(
Intensity(intensity_percent=intensity_percent),
wait_for_completion=True,
)
else:
await self.node.turn_on(wait_for_completion=True)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Instruct the light to turn off."""
await self.node.turn_off(wait_for_completion=True)
manifest.json
{
"domain": "velux",
"name": "Velux",
"documentation": "https://www.home-assistant.io/integrations/velux",
"requirements": ["pyvlx==0.2.20"],
"codeowners": ["@Julius2342"],
"iot_class": "local_polling",
"loggers": ["pyvlx"],
"version": "2022.1.28"
}
scene.py
"""Support for VELUX scenes."""
from __future__ import annotations
from typing import Any
from homeassistant.components.scene import Scene
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import _LOGGER, DATA_VELUX
PARALLEL_UPDATES = 1
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the scenes for Velux platform."""
entities = [VeluxScene(scene) for scene in hass.data[DATA_VELUX].pyvlx.scenes]
async_add_entities(entities)
class VeluxScene(Scene):
"""Representation of a Velux scene."""
def __init__(self, scene):
"""Init velux scene."""
_LOGGER.info("Adding Velux scene: %s", scene)
self.scene = scene
@property
def name(self):
"""Return the name of the scene."""
return self.scene.name
async def async_activate(self, **kwargs: Any) -> None:
"""Activate the scene."""
await self.scene.run(wait_for_completion=False)
services.yaml
# Velux Integration services
reboot_gateway:
strings.json
{
"services": {
"reboot_gateway": {
"name": "Reboot gateway",
"description": "Reboots the KLF200 Gateway."
}
}
}
The relevant different part is in cover.py
.at line 180ff.