Add cover support to PG LAB integration (#140290)

* Add cover support to PG LAB Electronics integration

* check shutter none state in is_closing and is_opening

* adding a loop instead of test test single cover individually
This commit is contained in:
pglab-electronics 2025-03-25 09:55:11 +01:00 committed by GitHub
parent 36d32eaabc
commit 13f306ddbc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 331 additions and 0 deletions

View File

@ -0,0 +1,107 @@
"""PG LAB Electronics Cover."""
from __future__ import annotations
from typing import Any
from pypglab.device import Device as PyPGLabDevice
from pypglab.shutter import Shutter as PyPGLabShutter
from homeassistant.components.cover import (
CoverDeviceClass,
CoverEntity,
CoverEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .discovery import PGLabDiscovery
from .entity import PGLabEntity
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up switches for device."""
@callback
def async_discover(
pglab_device: PyPGLabDevice, pglab_shutter: PyPGLabShutter
) -> None:
"""Discover and add a PG LAB Cover."""
pglab_discovery = config_entry.runtime_data
pglab_cover = PGLabCover(pglab_discovery, pglab_device, pglab_shutter)
async_add_entities([pglab_cover])
# Register the callback to create the cover entity when discovered.
pglab_discovery = config_entry.runtime_data
await pglab_discovery.register_platform(hass, Platform.COVER, async_discover)
class PGLabCover(PGLabEntity, CoverEntity):
"""A PGLab Cover."""
_attr_translation_key = "shutter"
def __init__(
self,
pglab_discovery: PGLabDiscovery,
pglab_device: PyPGLabDevice,
pglab_shutter: PyPGLabShutter,
) -> None:
"""Initialize the Cover class."""
super().__init__(
pglab_discovery,
pglab_device,
pglab_shutter,
)
self._attr_unique_id = f"{pglab_device.id}_shutter{pglab_shutter.id}"
self._attr_translation_placeholders = {"shutter_id": pglab_shutter.id}
self._shutter = pglab_shutter
self._attr_device_class = CoverDeviceClass.SHUTTER
self._attr_supported_features = (
CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE | CoverEntityFeature.STOP
)
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
await self._shutter.open()
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close cover."""
await self._shutter.close()
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
await self._shutter.stop()
@property
def is_closed(self) -> bool | None:
"""Return if cover is closed."""
if not self._shutter.state:
return None
return self._shutter.state == PyPGLabShutter.STATE_CLOSED
@property
def is_closing(self) -> bool | None:
"""Return if the cover is closing."""
if not self._shutter.state:
return None
return self._shutter.state == PyPGLabShutter.STATE_CLOSING
@property
def is_opening(self) -> bool | None:
"""Return if the cover is opening."""
if not self._shutter.state:
return None
return self._shutter.state == PyPGLabShutter.STATE_OPENING

View File

@ -34,12 +34,14 @@ if TYPE_CHECKING:
# Supported platforms.
PLATFORMS = [
Platform.COVER,
Platform.SENSOR,
Platform.SWITCH,
]
# Used to create a new component entity.
CREATE_NEW_ENTITY = {
Platform.COVER: "pglab_create_new_entity_cover",
Platform.SENSOR: "pglab_create_new_entity_sensor",
Platform.SWITCH: "pglab_create_new_entity_switch",
}
@ -250,6 +252,13 @@ class PGLabDiscovery:
)
self._discovered[pglab_device.id] = discovery_info
# Create all new cover entities.
for s in pglab_device.shutters:
# the HA entity is not yet created, send a message to create it
async_dispatcher_send(
hass, CREATE_NEW_ENTITY[Platform.COVER], pglab_device, s
)
# Create all new relay entities.
for r in pglab_device.relays:
# The HA entity is not yet created, send a message to create it.

View File

@ -15,6 +15,11 @@
}
},
"entity": {
"cover": {
"shutter": {
"name": "Shutter {shutter_id}"
}
},
"switch": {
"relay": {
"name": "Relay {relay_id}"

View File

@ -0,0 +1,210 @@
"""The tests for the PG LAB Electronics cover."""
import json
from homeassistant.components import cover
from homeassistant.components.cover import (
DOMAIN as COVER_DOMAIN,
SERVICE_CLOSE_COVER,
SERVICE_OPEN_COVER,
SERVICE_STOP_COVER,
)
from homeassistant.const import (
ATTR_ASSUMED_STATE,
STATE_CLOSED,
STATE_CLOSING,
STATE_OPEN,
STATE_OPENING,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from tests.common import async_fire_mqtt_message
from tests.typing import MqttMockHAClient
COVER_FEATURES = (
cover.CoverEntityFeature.OPEN
| cover.CoverEntityFeature.CLOSE
| cover.CoverEntityFeature.STOP
)
async def call_service(hass: HomeAssistant, entity_id, service, **kwargs):
"""Call a service."""
await hass.services.async_call(
COVER_DOMAIN,
service,
{"entity_id": entity_id, **kwargs},
blocking=True,
)
async def test_cover_features(
hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_pglab
) -> None:
"""Test cover features."""
topic = "pglab/discovery/E-Board-DD53AC85/config"
payload = {
"ip": "192.168.1.16",
"mac": "80:34:28:1B:18:5A",
"name": "test",
"hw": "1.0.7",
"fw": "1.0.0",
"type": "E-Board",
"id": "E-Board-DD53AC85",
"manufacturer": "PG LAB Electronics",
"params": {"shutters": 4, "boards": "10000000"},
}
async_fire_mqtt_message(
hass,
topic,
json.dumps(payload),
)
await hass.async_block_till_done()
assert len(hass.states.async_all("cover")) == 4
for i in range(4):
cover = hass.states.get(f"cover.test_shutter_{i}")
assert cover
assert cover.attributes["supported_features"] == COVER_FEATURES
async def test_cover_availability(
hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_pglab
) -> None:
"""Check if covers are properly created."""
topic = "pglab/discovery/E-Board-DD53AC85/config"
payload = {
"ip": "192.168.1.16",
"mac": "80:34:28:1B:18:5A",
"name": "test",
"hw": "1.0.7",
"fw": "1.0.0",
"type": "E-Board",
"id": "E-Board-DD53AC85",
"manufacturer": "PG LAB Electronics",
"params": {"shutters": 6, "boards": "11000000"},
}
async_fire_mqtt_message(
hass,
topic,
json.dumps(payload),
)
await hass.async_block_till_done()
# We are creating 6 covers using two E-RELAY devices connected to E-BOARD.
# Now we are going to check if all covers are created and their state is unknown.
for i in range(5):
cover = hass.states.get(f"cover.test_shutter_{i}")
assert cover.state == STATE_UNKNOWN
assert not cover.attributes.get(ATTR_ASSUMED_STATE)
# The cover with id 7 should not be created.
cover = hass.states.get("cover.test_shutter_7")
assert not cover
async def test_cover_change_state_via_mqtt(
hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_pglab
) -> None:
"""Test state update via MQTT."""
topic = "pglab/discovery/E-Board-DD53AC85/config"
payload = {
"ip": "192.168.1.16",
"mac": "80:34:28:1B:18:5A",
"name": "test",
"hw": "1.0.7",
"fw": "1.0.0",
"type": "E-Board",
"id": "E-Board-DD53AC85",
"manufacturer": "PG LAB Electronics",
"params": {"shutters": 2, "boards": "10000000"},
}
async_fire_mqtt_message(
hass,
topic,
json.dumps(payload),
)
await hass.async_block_till_done()
# Check initial state is unknown
cover = hass.states.get("cover.test_shutter_0")
assert cover.state == STATE_UNKNOWN
assert not cover.attributes.get(ATTR_ASSUMED_STATE)
# Simulate the device responds sending mqtt messages and check if the cover state
# change appropriately.
async_fire_mqtt_message(hass, "pglab/test/shutter/0/state", "OPEN")
await hass.async_block_till_done()
cover = hass.states.get("cover.test_shutter_0")
assert not cover.attributes.get(ATTR_ASSUMED_STATE)
assert cover.state == STATE_OPEN
async_fire_mqtt_message(hass, "pglab/test/shutter/0/state", "OPENING")
await hass.async_block_till_done()
cover = hass.states.get("cover.test_shutter_0")
assert cover.state == STATE_OPENING
async_fire_mqtt_message(hass, "pglab/test/shutter/0/state", "CLOSING")
await hass.async_block_till_done()
cover = hass.states.get("cover.test_shutter_0")
assert cover.state == STATE_CLOSING
async_fire_mqtt_message(hass, "pglab/test/shutter/0/state", "CLOSED")
await hass.async_block_till_done()
cover = hass.states.get("cover.test_shutter_0")
assert cover.state == STATE_CLOSED
async def test_cover_mqtt_state_by_calling_service(
hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_pglab
) -> None:
"""Calling service to OPEN/CLOSE cover and check mqtt state."""
topic = "pglab/discovery/E-Board-DD53AC85/config"
payload = {
"ip": "192.168.1.16",
"mac": "80:34:28:1B:18:5A",
"name": "test",
"hw": "1.0.7",
"fw": "1.0.0",
"type": "E-Board",
"id": "E-Board-DD53AC85",
"manufacturer": "PG LAB Electronics",
"params": {"shutters": 2, "boards": "10000000"},
}
async_fire_mqtt_message(
hass,
topic,
json.dumps(payload),
)
await hass.async_block_till_done()
cover = hass.states.get("cover.test_shutter_0")
assert cover.state == STATE_UNKNOWN
assert not cover.attributes.get(ATTR_ASSUMED_STATE)
# Call HA covers services and verify that the MQTT messages are sent correctly
await call_service(hass, "cover.test_shutter_0", SERVICE_OPEN_COVER)
mqtt_mock.async_publish.assert_called_once_with(
"pglab/test/shutter/0/set", "OPEN", 0, False
)
mqtt_mock.async_publish.reset_mock()
await call_service(hass, "cover.test_shutter_0", SERVICE_STOP_COVER)
mqtt_mock.async_publish.assert_called_once_with(
"pglab/test/shutter/0/set", "STOP", 0, False
)
mqtt_mock.async_publish.reset_mock()
await call_service(hass, "cover.test_shutter_0", SERVICE_CLOSE_COVER)
mqtt_mock.async_publish.assert_called_once_with(
"pglab/test/shutter/0/set", "CLOSE", 0, False
)
mqtt_mock.async_publish.reset_mock()