Improve code quality telnet (#65239)

This commit is contained in:
G Johansson 2022-02-12 15:15:28 +01:00 committed by GitHub
parent 8da150bd71
commit db6969739f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23

View file

@ -4,6 +4,7 @@ from __future__ import annotations
from datetime import timedelta from datetime import timedelta
import logging import logging
import telnetlib import telnetlib
from typing import Any
import voluptuous as vol import voluptuous as vol
@ -26,6 +27,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -60,27 +62,26 @@ def setup_platform(
discovery_info: DiscoveryInfoType | None = None, discovery_info: DiscoveryInfoType | None = None,
) -> None: ) -> None:
"""Find and return switches controlled by telnet commands.""" """Find and return switches controlled by telnet commands."""
devices = config.get(CONF_SWITCHES, {}) devices: dict[str, Any] = config[CONF_SWITCHES]
switches = [] switches = []
for object_id, device_config in devices.items(): for object_id, device_config in devices.items():
value_template = device_config.get(CONF_VALUE_TEMPLATE) value_template: Template | None = device_config.get(CONF_VALUE_TEMPLATE)
if value_template is not None: if value_template is not None:
value_template.hass = hass value_template.hass = hass
switches.append( switches.append(
TelnetSwitch( TelnetSwitch(
hass,
object_id, object_id,
device_config.get(CONF_RESOURCE), device_config[CONF_RESOURCE],
device_config.get(CONF_PORT), device_config[CONF_PORT],
device_config.get(CONF_NAME, object_id), device_config.get(CONF_NAME, object_id),
device_config.get(CONF_COMMAND_ON), device_config[CONF_COMMAND_ON],
device_config.get(CONF_COMMAND_OFF), device_config[CONF_COMMAND_OFF],
device_config.get(CONF_COMMAND_STATE), device_config.get(CONF_COMMAND_STATE),
value_template, value_template,
device_config.get(CONF_TIMEOUT), device_config[CONF_TIMEOUT],
) )
) )
@ -96,82 +97,65 @@ class TelnetSwitch(SwitchEntity):
def __init__( def __init__(
self, self,
hass, object_id: str,
object_id, resource: str,
resource, port: int,
port, friendly_name: str,
friendly_name, command_on: str,
command_on, command_off: str,
command_off, command_state: str | None,
command_state, value_template: Template | None,
value_template, timeout: float,
timeout, ) -> None:
):
"""Initialize the switch.""" """Initialize the switch."""
self._hass = hass
self.entity_id = ENTITY_ID_FORMAT.format(object_id) self.entity_id = ENTITY_ID_FORMAT.format(object_id)
self._resource = resource self._resource = resource
self._port = port self._port = port
self._name = friendly_name self._attr_name = friendly_name
self._state = False self._attr_is_on = False
self._command_on = command_on self._command_on = command_on
self._command_off = command_off self._command_off = command_off
self._command_state = command_state self._command_state = command_state
self._value_template = value_template self._value_template = value_template
self._timeout = timeout self._timeout = timeout
self._attr_should_poll = bool(command_state)
self._attr_assumed_state = bool(command_state is None)
def _telnet_command(self, command): def _telnet_command(self, command: str) -> str | None:
try: try:
telnet = telnetlib.Telnet(self._resource, self._port) telnet = telnetlib.Telnet(self._resource, self._port)
telnet.write(command.encode("ASCII") + b"\r") telnet.write(command.encode("ASCII") + b"\r")
response = telnet.read_until(b"\r", timeout=self._timeout) response = telnet.read_until(b"\r", timeout=self._timeout)
_LOGGER.debug("telnet response: %s", response.decode("ASCII").strip())
return response.decode("ASCII").strip()
except OSError as error: except OSError as error:
_LOGGER.error( _LOGGER.error(
'Command "%s" failed with exception: %s', command, repr(error) 'Command "%s" failed with exception: %s', command, repr(error)
) )
return None return None
_LOGGER.debug("telnet response: %s", response.decode("ASCII").strip())
return response.decode("ASCII").strip()
@property def update(self) -> None:
def name(self):
"""Return the name of the switch."""
return self._name
@property
def should_poll(self):
"""Only poll if we have state command."""
return self._command_state is not None
@property
def is_on(self):
"""Return true if device is on."""
return self._state
@property
def assumed_state(self):
"""Return true if no state command is defined, false otherwise."""
return self._command_state is None
def update(self):
"""Update device state.""" """Update device state."""
if not self._command_state:
return
response = self._telnet_command(self._command_state) response = self._telnet_command(self._command_state)
if response: if response and self._value_template:
rendered = self._value_template.render_with_possible_json_value(response) rendered = self._value_template.render_with_possible_json_value(response)
self._state = rendered == "True"
else: else:
_LOGGER.warning("Empty response for command: %s", self._command_state) _LOGGER.warning("Empty response for command: %s", self._command_state)
return None
self._attr_is_on = rendered == "True"
def turn_on(self, **kwargs): def turn_on(self, **kwargs) -> None:
"""Turn the device on.""" """Turn the device on."""
self._telnet_command(self._command_on) self._telnet_command(self._command_on)
if self.assumed_state: if self.assumed_state:
self._state = True self._attr_is_on = True
self.schedule_update_ha_state() self.schedule_update_ha_state()
def turn_off(self, **kwargs): def turn_off(self, **kwargs) -> None:
"""Turn the device off.""" """Turn the device off."""
self._telnet_command(self._command_off) self._telnet_command(self._command_off)
if self.assumed_state: if self.assumed_state:
self._state = False self._attr_is_on = False
self.schedule_update_ha_state() self.schedule_update_ha_state()