Add strict type annotations to tcp (#50877)
* add strict type annotations * apply suggestions * rename to TCP_PLATFORM_SCHEMA * Replace DiscoveryInfoTypepull/50964/head
parent
560dd0a0cc
commit
4a64f7a696
|
@ -59,6 +59,7 @@ homeassistant.components.sun.*
|
|||
homeassistant.components.switch.*
|
||||
homeassistant.components.synology_dsm.*
|
||||
homeassistant.components.systemmonitor.*
|
||||
homeassistant.components.tcp.*
|
||||
homeassistant.components.tts.*
|
||||
homeassistant.components.upcloud.*
|
||||
homeassistant.components.vacuum.*
|
||||
|
|
|
@ -1,12 +1,25 @@
|
|||
"""Provides a binary sensor which gets its values from a TCP socket."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Final
|
||||
|
||||
from homeassistant.components.binary_sensor import BinarySensorEntity
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from .sensor import CONF_VALUE_ON, PLATFORM_SCHEMA, TcpSensor
|
||||
from .const import CONF_VALUE_ON
|
||||
from .sensor import PLATFORM_SCHEMA as TCP_PLATFORM_SCHEMA, TcpSensor
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({})
|
||||
PLATFORM_SCHEMA: Final = TCP_PLATFORM_SCHEMA
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
def setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
add_entities: AddEntitiesCallback,
|
||||
discovery_info: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Set up the TCP binary sensor."""
|
||||
add_entities([TcpBinarySensor(hass, config)])
|
||||
|
||||
|
@ -14,9 +27,7 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
|
|||
class TcpBinarySensor(BinarySensorEntity, TcpSensor):
|
||||
"""A binary sensor which is on when its state == CONF_VALUE_ON."""
|
||||
|
||||
required = (CONF_VALUE_ON,)
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if the binary sensor is on."""
|
||||
return self._state == self._config[CONF_VALUE_ON]
|
||||
|
|
|
@ -0,0 +1,13 @@
|
|||
"""Constants for TCP platform."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Final
|
||||
|
||||
CONF_BUFFER_SIZE: Final = "buffer_size"
|
||||
CONF_VALUE_ON: Final = "value_on"
|
||||
|
||||
DEFAULT_BUFFER_SIZE: Final = 1024
|
||||
DEFAULT_NAME: Final = "TCP Sensor"
|
||||
DEFAULT_TIMEOUT: Final = 10
|
||||
DEFAULT_SSL: Final = False
|
||||
DEFAULT_VERIFY_SSL: Final = True
|
|
@ -0,0 +1,22 @@
|
|||
"""Models for TCP platform."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TypedDict
|
||||
|
||||
from homeassistant.helpers.template import Template
|
||||
|
||||
|
||||
class TcpSensorConfig(TypedDict):
|
||||
"""TypedDict for TcpSensor config."""
|
||||
|
||||
name: str
|
||||
host: str
|
||||
port: str
|
||||
timeout: int
|
||||
payload: str
|
||||
unit_of_measurement: str | None
|
||||
value_template: Template | None
|
||||
value_on: str | None
|
||||
buffer_size: int
|
||||
ssl: bool
|
||||
verify_ssl: bool
|
|
@ -1,12 +1,18 @@
|
|||
"""Support for TCP socket based sensors."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import select
|
||||
import socket
|
||||
import ssl
|
||||
from typing import Any, Final
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
|
||||
from homeassistant.components.sensor import (
|
||||
PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
|
||||
SensorEntity,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_NAME,
|
||||
|
@ -18,21 +24,27 @@ from homeassistant.const import (
|
|||
CONF_VALUE_TEMPLATE,
|
||||
CONF_VERIFY_SSL,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import TemplateError
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.template import Template
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
from .const import (
|
||||
CONF_BUFFER_SIZE,
|
||||
CONF_VALUE_ON,
|
||||
DEFAULT_BUFFER_SIZE,
|
||||
DEFAULT_NAME,
|
||||
DEFAULT_SSL,
|
||||
DEFAULT_TIMEOUT,
|
||||
DEFAULT_VERIFY_SSL,
|
||||
)
|
||||
from .model import TcpSensorConfig
|
||||
|
||||
CONF_BUFFER_SIZE = "buffer_size"
|
||||
CONF_VALUE_ON = "value_on"
|
||||
_LOGGER: Final = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_BUFFER_SIZE = 1024
|
||||
DEFAULT_NAME = "TCP Sensor"
|
||||
DEFAULT_TIMEOUT = 10
|
||||
DEFAULT_SSL = False
|
||||
DEFAULT_VERIFY_SSL = True
|
||||
|
||||
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
||||
PLATFORM_SCHEMA: Final = PARENT_PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_HOST): cv.string,
|
||||
vol.Required(CONF_PORT): cv.port,
|
||||
|
@ -49,7 +61,12 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
|
|||
)
|
||||
|
||||
|
||||
def setup_platform(hass, config, add_entities, discovery_info=None):
|
||||
def setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
add_entities: AddEntitiesCallback,
|
||||
discovery_info: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Set up the TCP Sensor."""
|
||||
add_entities([TcpSensor(hass, config)])
|
||||
|
||||
|
@ -57,55 +74,54 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
|
|||
class TcpSensor(SensorEntity):
|
||||
"""Implementation of a TCP socket based sensor."""
|
||||
|
||||
required = ()
|
||||
|
||||
def __init__(self, hass, config):
|
||||
def __init__(self, hass: HomeAssistant, config: ConfigType) -> None:
|
||||
"""Set all the config values if they exist and get initial state."""
|
||||
value_template = config.get(CONF_VALUE_TEMPLATE)
|
||||
|
||||
value_template: Template | None = config.get(CONF_VALUE_TEMPLATE)
|
||||
if value_template is not None:
|
||||
value_template.hass = hass
|
||||
|
||||
self._hass = hass
|
||||
self._config = {
|
||||
CONF_NAME: config.get(CONF_NAME),
|
||||
CONF_HOST: config.get(CONF_HOST),
|
||||
CONF_PORT: config.get(CONF_PORT),
|
||||
CONF_TIMEOUT: config.get(CONF_TIMEOUT),
|
||||
CONF_PAYLOAD: config.get(CONF_PAYLOAD),
|
||||
self._config: TcpSensorConfig = {
|
||||
CONF_NAME: config[CONF_NAME],
|
||||
CONF_HOST: config[CONF_HOST],
|
||||
CONF_PORT: config[CONF_PORT],
|
||||
CONF_TIMEOUT: config[CONF_TIMEOUT],
|
||||
CONF_PAYLOAD: config[CONF_PAYLOAD],
|
||||
CONF_UNIT_OF_MEASUREMENT: config.get(CONF_UNIT_OF_MEASUREMENT),
|
||||
CONF_VALUE_TEMPLATE: value_template,
|
||||
CONF_VALUE_ON: config.get(CONF_VALUE_ON),
|
||||
CONF_BUFFER_SIZE: config.get(CONF_BUFFER_SIZE),
|
||||
CONF_BUFFER_SIZE: config[CONF_BUFFER_SIZE],
|
||||
CONF_SSL: config[CONF_SSL],
|
||||
CONF_VERIFY_SSL: config[CONF_VERIFY_SSL],
|
||||
}
|
||||
|
||||
if config[CONF_SSL]:
|
||||
self._ssl_context: ssl.SSLContext | None = None
|
||||
if self._config[CONF_SSL]:
|
||||
self._ssl_context = ssl.create_default_context()
|
||||
if not config[CONF_VERIFY_SSL]:
|
||||
if not self._config[CONF_VERIFY_SSL]:
|
||||
self._ssl_context.check_hostname = False
|
||||
self._ssl_context.verify_mode = ssl.CERT_NONE
|
||||
else:
|
||||
self._ssl_context = None
|
||||
|
||||
self._state = None
|
||||
self._state: str | None = None
|
||||
self.update()
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
def name(self) -> str:
|
||||
"""Return the name of this sensor."""
|
||||
return self._config[CONF_NAME]
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
def state(self) -> str | None:
|
||||
"""Return the state of the device."""
|
||||
return self._state
|
||||
|
||||
@property
|
||||
def unit_of_measurement(self):
|
||||
def unit_of_measurement(self) -> str | None:
|
||||
"""Return the unit of measurement of this entity."""
|
||||
return self._config[CONF_UNIT_OF_MEASUREMENT]
|
||||
|
||||
def update(self):
|
||||
def update(self) -> None:
|
||||
"""Get the latest value for this sensor."""
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||||
sock.settimeout(self._config[CONF_TIMEOUT])
|
||||
|
@ -151,11 +167,10 @@ class TcpSensor(SensorEntity):
|
|||
|
||||
value = sock.recv(self._config[CONF_BUFFER_SIZE]).decode()
|
||||
|
||||
if self._config[CONF_VALUE_TEMPLATE] is not None:
|
||||
value_template = self._config[CONF_VALUE_TEMPLATE]
|
||||
if value_template is not None:
|
||||
try:
|
||||
self._state = self._config[CONF_VALUE_TEMPLATE].render(
|
||||
parse_result=False, value=value
|
||||
)
|
||||
self._state = value_template.render(parse_result=False, value=value)
|
||||
return
|
||||
except TemplateError:
|
||||
_LOGGER.error(
|
||||
|
|
14
mypy.ini
14
mypy.ini
|
@ -660,6 +660,17 @@ no_implicit_optional = true
|
|||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.tcp.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
no_implicit_optional = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.tts.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
@ -1331,9 +1342,6 @@ ignore_errors = true
|
|||
[mypy-homeassistant.components.tasmota.*]
|
||||
ignore_errors = true
|
||||
|
||||
[mypy-homeassistant.components.tcp.*]
|
||||
ignore_errors = true
|
||||
|
||||
[mypy-homeassistant.components.telegram_bot.*]
|
||||
ignore_errors = true
|
||||
|
||||
|
|
|
@ -201,7 +201,6 @@ IGNORED_MODULES: Final[list[str]] = [
|
|||
"homeassistant.components.system_log.*",
|
||||
"homeassistant.components.tado.*",
|
||||
"homeassistant.components.tasmota.*",
|
||||
"homeassistant.components.tcp.*",
|
||||
"homeassistant.components.telegram_bot.*",
|
||||
"homeassistant.components.template.*",
|
||||
"homeassistant.components.tesla.*",
|
||||
|
|
Loading…
Reference in New Issue