Add support for Broadlink BG1 devices (#42314)

* Support for BG1 switches after config flow updates to core Broadlink component

* updates based on @felipediel feedback

* Update updater.py

* Update switch.py

* Update switch.py
This commit is contained in:
b4dpxl 2020-11-14 13:59:41 +00:00 committed by GitHub
parent cc396b9736
commit bb31de1de7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 57 additions and 1 deletions

View File

@ -8,7 +8,7 @@ DOMAIN = "broadlink"
DOMAINS_AND_TYPES = (
(REMOTE_DOMAIN, ("RM2", "RM4")),
(SENSOR_DOMAIN, ("A1", "RM2", "RM4")),
(SWITCH_DOMAIN, ("MP1", "RM2", "RM4", "SP1", "SP2", "SP4", "SP4B")),
(SWITCH_DOMAIN, ("BG1", "MP1", "RM2", "RM4", "SP1", "SP2", "SP4", "SP4B")),
)
DEFAULT_PORT = 80

View File

@ -1,5 +1,6 @@
"""Support for Broadlink switches."""
from abc import ABC, abstractmethod
from functools import partial
import logging
from broadlink.exceptions import BroadlinkException
@ -124,6 +125,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
elif device.api.type in {"SP4", "SP4B"}:
switches = [BroadlinkSP4Switch(device)]
elif device.api.type == "BG1":
switches = [BroadlinkBG1Slot(device, slot) for slot in range(1, 3)]
elif device.api.type == "MP1":
switches = [BroadlinkMP1Slot(device, slot) for slot in range(1, 5)]
@ -360,3 +364,46 @@ class BroadlinkMP1Slot(BroadlinkSwitch):
_LOGGER.error("Failed to send packet: %s", err)
return False
return True
class BroadlinkBG1Slot(BroadlinkSwitch):
"""Representation of a Broadlink BG1 slot."""
def __init__(self, device, slot):
"""Initialize the switch."""
super().__init__(device, 1, 0)
self._slot = slot
self._state = self._coordinator.data[f"pwr{slot}"]
self._device_class = DEVICE_CLASS_OUTLET
@property
def unique_id(self):
"""Return the unique id of the slot."""
return f"{self._device.unique_id}-s{self._slot}"
@property
def name(self):
"""Return the name of the switch."""
return f"{self._device.name} S{self._slot}"
@property
def assumed_state(self):
"""Return True if unable to access real state of the switch."""
return False
@callback
def update_data(self):
"""Update data."""
if self._coordinator.last_update_success:
self._state = self._coordinator.data[f"pwr{self._slot}"]
self.async_write_ha_state()
async def _async_send_packet(self, packet):
"""Send a packet to the device."""
set_state = partial(self._device.api.set_state, **{f"pwr{self._slot}": packet})
try:
await self._device.async_request(set_state)
except (BroadlinkException, OSError) as err:
_LOGGER.error("Failed to send packet: %s", err)
return False
return True

View File

@ -26,6 +26,7 @@ def get_update_manager(device):
update_managers = {
"A1": BroadlinkA1UpdateManager,
"BG1": BroadlinkBG1UpdateManager,
"MP1": BroadlinkMP1UpdateManager,
"RM2": BroadlinkRMUpdateManager,
"RM4": BroadlinkRMUpdateManager,
@ -161,6 +162,14 @@ class BroadlinkSP2UpdateManager(BroadlinkUpdateManager):
return data
class BroadlinkBG1UpdateManager(BroadlinkUpdateManager):
"""Manages updates for Broadlink BG1 devices."""
async def async_fetch_data(self):
"""Fetch data from the device."""
return await self.device.async_request(self.device.api.get_state)
class BroadlinkSP4UpdateManager(BroadlinkUpdateManager):
"""Manages updates for Broadlink SP4 devices."""