mirror of
https://github.com/home-assistant/core.git
synced 2025-04-24 09:17:53 +00:00
Cleanup ADS constants and add type hints (#63390)
Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
parent
55143e240b
commit
0042bb68d9
@ -16,9 +16,10 @@ from homeassistant.const import (
|
||||
CONF_PORT,
|
||||
EVENT_HOMEASSISTANT_STOP,
|
||||
)
|
||||
from homeassistant.core import ServiceCall
|
||||
from homeassistant.core import HomeAssistant, ServiceCall
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@ -32,6 +33,15 @@ ADSTYPE_INT = "int"
|
||||
ADSTYPE_UDINT = "udint"
|
||||
ADSTYPE_UINT = "uint"
|
||||
|
||||
ADS_TYPEMAP = {
|
||||
ADSTYPE_BOOL: pyads.PLCTYPE_BOOL,
|
||||
ADSTYPE_BYTE: pyads.PLCTYPE_BYTE,
|
||||
ADSTYPE_DINT: pyads.PLCTYPE_DINT,
|
||||
ADSTYPE_INT: pyads.PLCTYPE_INT,
|
||||
ADSTYPE_UDINT: pyads.PLCTYPE_UDINT,
|
||||
ADSTYPE_UINT: pyads.PLCTYPE_UINT,
|
||||
}
|
||||
|
||||
CONF_ADS_FACTOR = "factor"
|
||||
CONF_ADS_TYPE = "adstype"
|
||||
CONF_ADS_VALUE = "value"
|
||||
@ -78,7 +88,7 @@ SCHEMA_SERVICE_WRITE_DATA_BY_NAME = vol.Schema(
|
||||
)
|
||||
|
||||
|
||||
def setup(hass, config):
|
||||
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the ADS component."""
|
||||
|
||||
conf = config[DOMAIN]
|
||||
@ -89,23 +99,6 @@ def setup(hass, config):
|
||||
|
||||
client = pyads.Connection(net_id, port, ip_address)
|
||||
|
||||
AdsHub.ADS_TYPEMAP = {
|
||||
ADSTYPE_BOOL: pyads.PLCTYPE_BOOL,
|
||||
ADSTYPE_BYTE: pyads.PLCTYPE_BYTE,
|
||||
ADSTYPE_DINT: pyads.PLCTYPE_DINT,
|
||||
ADSTYPE_INT: pyads.PLCTYPE_INT,
|
||||
ADSTYPE_UDINT: pyads.PLCTYPE_UDINT,
|
||||
ADSTYPE_UINT: pyads.PLCTYPE_UINT,
|
||||
}
|
||||
|
||||
AdsHub.ADSError = pyads.ADSError
|
||||
AdsHub.PLCTYPE_BOOL = pyads.PLCTYPE_BOOL
|
||||
AdsHub.PLCTYPE_BYTE = pyads.PLCTYPE_BYTE
|
||||
AdsHub.PLCTYPE_DINT = pyads.PLCTYPE_DINT
|
||||
AdsHub.PLCTYPE_INT = pyads.PLCTYPE_INT
|
||||
AdsHub.PLCTYPE_UDINT = pyads.PLCTYPE_UDINT
|
||||
AdsHub.PLCTYPE_UINT = pyads.PLCTYPE_UINT
|
||||
|
||||
try:
|
||||
ads = AdsHub(client)
|
||||
except pyads.ADSError:
|
||||
@ -122,12 +115,12 @@ def setup(hass, config):
|
||||
|
||||
def handle_write_data_by_name(call: ServiceCall) -> None:
|
||||
"""Write a value to the connected ADS device."""
|
||||
ads_var = call.data.get(CONF_ADS_VAR)
|
||||
ads_type = call.data.get(CONF_ADS_TYPE)
|
||||
value = call.data.get(CONF_ADS_VALUE)
|
||||
ads_var = call.data[CONF_ADS_VAR]
|
||||
ads_type = call.data[CONF_ADS_TYPE]
|
||||
value = call.data[CONF_ADS_VALUE]
|
||||
|
||||
try:
|
||||
ads.write_by_name(ads_var, value, ads.ADS_TYPEMAP[ads_type])
|
||||
ads.write_by_name(ads_var, value, ADS_TYPEMAP[ads_type])
|
||||
except pyads.ADSError as err:
|
||||
_LOGGER.error(err)
|
||||
|
||||
@ -247,17 +240,17 @@ class AdsHub:
|
||||
return
|
||||
|
||||
# Parse data to desired datatype
|
||||
if notification_item.plc_datatype == self.PLCTYPE_BOOL:
|
||||
if notification_item.plc_datatype == pyads.PLCTYPE_BOOL:
|
||||
value = bool(struct.unpack("<?", bytearray(data))[0])
|
||||
elif notification_item.plc_datatype == self.PLCTYPE_INT:
|
||||
elif notification_item.plc_datatype == pyads.PLCTYPE_INT:
|
||||
value = struct.unpack("<h", bytearray(data))[0]
|
||||
elif notification_item.plc_datatype == self.PLCTYPE_BYTE:
|
||||
elif notification_item.plc_datatype == pyads.PLCTYPE_BYTE:
|
||||
value = struct.unpack("<B", bytearray(data))[0]
|
||||
elif notification_item.plc_datatype == self.PLCTYPE_UINT:
|
||||
elif notification_item.plc_datatype == pyads.PLCTYPE_UINT:
|
||||
value = struct.unpack("<H", bytearray(data))[0]
|
||||
elif notification_item.plc_datatype == self.PLCTYPE_DINT:
|
||||
elif notification_item.plc_datatype == pyads.PLCTYPE_DINT:
|
||||
value = struct.unpack("<i", bytearray(data))[0]
|
||||
elif notification_item.plc_datatype == self.PLCTYPE_UDINT:
|
||||
elif notification_item.plc_datatype == pyads.PLCTYPE_UDINT:
|
||||
value = struct.unpack("<I", bytearray(data))[0]
|
||||
else:
|
||||
value = bytearray(data)
|
||||
|
@ -1,6 +1,7 @@
|
||||
"""Support for ADS binary sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
import pyads
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.binary_sensor import (
|
||||
@ -54,7 +55,7 @@ class AdsBinarySensor(AdsEntity, BinarySensorEntity):
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Register device notification."""
|
||||
await self.async_initialize_device(self._ads_var, self._ads_hub.PLCTYPE_BOOL)
|
||||
await self.async_initialize_device(self._ads_var, pyads.PLCTYPE_BOOL)
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool:
|
||||
|
@ -1,6 +1,7 @@
|
||||
"""Support for ADS covers."""
|
||||
from __future__ import annotations
|
||||
|
||||
import pyads
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.cover import (
|
||||
@ -125,13 +126,11 @@ class AdsCover(AdsEntity, CoverEntity):
|
||||
async def async_added_to_hass(self):
|
||||
"""Register device notification."""
|
||||
if self._ads_var is not None:
|
||||
await self.async_initialize_device(
|
||||
self._ads_var, self._ads_hub.PLCTYPE_BOOL
|
||||
)
|
||||
await self.async_initialize_device(self._ads_var, pyads.PLCTYPE_BOOL)
|
||||
|
||||
if self._ads_var_position is not None:
|
||||
await self.async_initialize_device(
|
||||
self._ads_var_position, self._ads_hub.PLCTYPE_BYTE, STATE_KEY_POSITION
|
||||
self._ads_var_position, pyads.PLCTYPE_BYTE, STATE_KEY_POSITION
|
||||
)
|
||||
|
||||
@property
|
||||
@ -151,33 +150,27 @@ class AdsCover(AdsEntity, CoverEntity):
|
||||
def stop_cover(self, **kwargs):
|
||||
"""Fire the stop action."""
|
||||
if self._ads_var_stop:
|
||||
self._ads_hub.write_by_name(
|
||||
self._ads_var_stop, True, self._ads_hub.PLCTYPE_BOOL
|
||||
)
|
||||
self._ads_hub.write_by_name(self._ads_var_stop, True, pyads.PLCTYPE_BOOL)
|
||||
|
||||
def set_cover_position(self, **kwargs):
|
||||
"""Set cover position."""
|
||||
position = kwargs[ATTR_POSITION]
|
||||
if self._ads_var_pos_set is not None:
|
||||
self._ads_hub.write_by_name(
|
||||
self._ads_var_pos_set, position, self._ads_hub.PLCTYPE_BYTE
|
||||
self._ads_var_pos_set, position, pyads.PLCTYPE_BYTE
|
||||
)
|
||||
|
||||
def open_cover(self, **kwargs):
|
||||
"""Move the cover up."""
|
||||
if self._ads_var_open is not None:
|
||||
self._ads_hub.write_by_name(
|
||||
self._ads_var_open, True, self._ads_hub.PLCTYPE_BOOL
|
||||
)
|
||||
self._ads_hub.write_by_name(self._ads_var_open, True, pyads.PLCTYPE_BOOL)
|
||||
elif self._ads_var_pos_set is not None:
|
||||
self.set_cover_position(position=100)
|
||||
|
||||
def close_cover(self, **kwargs):
|
||||
"""Move the cover down."""
|
||||
if self._ads_var_close is not None:
|
||||
self._ads_hub.write_by_name(
|
||||
self._ads_var_close, True, self._ads_hub.PLCTYPE_BOOL
|
||||
)
|
||||
self._ads_hub.write_by_name(self._ads_var_close, True, pyads.PLCTYPE_BOOL)
|
||||
elif self._ads_var_pos_set is not None:
|
||||
self.set_cover_position(position=0)
|
||||
|
||||
|
@ -1,6 +1,7 @@
|
||||
"""Support for ADS light sources."""
|
||||
from __future__ import annotations
|
||||
|
||||
import pyads
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.light import (
|
||||
@ -63,12 +64,12 @@ class AdsLight(AdsEntity, LightEntity):
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Register device notification."""
|
||||
await self.async_initialize_device(self._ads_var, self._ads_hub.PLCTYPE_BOOL)
|
||||
await self.async_initialize_device(self._ads_var, pyads.PLCTYPE_BOOL)
|
||||
|
||||
if self._ads_var_brightness is not None:
|
||||
await self.async_initialize_device(
|
||||
self._ads_var_brightness,
|
||||
self._ads_hub.PLCTYPE_UINT,
|
||||
pyads.PLCTYPE_UINT,
|
||||
STATE_KEY_BRIGHTNESS,
|
||||
)
|
||||
|
||||
@ -85,13 +86,13 @@ class AdsLight(AdsEntity, LightEntity):
|
||||
def turn_on(self, **kwargs):
|
||||
"""Turn the light on or set a specific dimmer value."""
|
||||
brightness = kwargs.get(ATTR_BRIGHTNESS)
|
||||
self._ads_hub.write_by_name(self._ads_var, True, self._ads_hub.PLCTYPE_BOOL)
|
||||
self._ads_hub.write_by_name(self._ads_var, True, pyads.PLCTYPE_BOOL)
|
||||
|
||||
if self._ads_var_brightness is not None and brightness is not None:
|
||||
self._ads_hub.write_by_name(
|
||||
self._ads_var_brightness, brightness, self._ads_hub.PLCTYPE_UINT
|
||||
self._ads_var_brightness, brightness, pyads.PLCTYPE_UINT
|
||||
)
|
||||
|
||||
def turn_off(self, **kwargs):
|
||||
"""Turn the light off."""
|
||||
self._ads_hub.write_by_name(self._ads_var, False, self._ads_hub.PLCTYPE_BOOL)
|
||||
self._ads_hub.write_by_name(self._ads_var, False, pyads.PLCTYPE_BOOL)
|
||||
|
@ -11,7 +11,14 @@ import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateType
|
||||
|
||||
from . import CONF_ADS_FACTOR, CONF_ADS_TYPE, CONF_ADS_VAR, STATE_KEY_STATE, AdsEntity
|
||||
from . import (
|
||||
ADS_TYPEMAP,
|
||||
CONF_ADS_FACTOR,
|
||||
CONF_ADS_TYPE,
|
||||
CONF_ADS_VAR,
|
||||
STATE_KEY_STATE,
|
||||
AdsEntity,
|
||||
)
|
||||
|
||||
DEFAULT_NAME = "ADS sensor"
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
@ -67,7 +74,7 @@ class AdsSensor(AdsEntity, SensorEntity):
|
||||
"""Register device notification."""
|
||||
await self.async_initialize_device(
|
||||
self._ads_var,
|
||||
self._ads_hub.ADS_TYPEMAP[self._ads_type],
|
||||
ADS_TYPEMAP[self._ads_type],
|
||||
STATE_KEY_STATE,
|
||||
self._factor,
|
||||
)
|
||||
|
@ -1,6 +1,7 @@
|
||||
"""Support for ADS switch platform."""
|
||||
from __future__ import annotations
|
||||
|
||||
import pyads
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.switch import PLATFORM_SCHEMA, SwitchEntity
|
||||
@ -42,7 +43,7 @@ class AdsSwitch(AdsEntity, SwitchEntity):
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Register device notification."""
|
||||
await self.async_initialize_device(self._ads_var, self._ads_hub.PLCTYPE_BOOL)
|
||||
await self.async_initialize_device(self._ads_var, pyads.PLCTYPE_BOOL)
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool:
|
||||
@ -51,8 +52,8 @@ class AdsSwitch(AdsEntity, SwitchEntity):
|
||||
|
||||
def turn_on(self, **kwargs):
|
||||
"""Turn the switch on."""
|
||||
self._ads_hub.write_by_name(self._ads_var, True, self._ads_hub.PLCTYPE_BOOL)
|
||||
self._ads_hub.write_by_name(self._ads_var, True, pyads.PLCTYPE_BOOL)
|
||||
|
||||
def turn_off(self, **kwargs):
|
||||
"""Turn the switch off."""
|
||||
self._ads_hub.write_by_name(self._ads_var, False, self._ads_hub.PLCTYPE_BOOL)
|
||||
self._ads_hub.write_by_name(self._ads_var, False, pyads.PLCTYPE_BOOL)
|
||||
|
Loading…
x
Reference in New Issue
Block a user