evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Always save current position if payload is numeric value (#16148)

Browse source 
* Change of behaviour. Allow user to configure either a position topic or a state topic but not
both.

* optimistic mode in set_cover and tests added

* optimistic mode in set_cover_position using percentage_position

* fixes accroding to Martin review.

* added validation schema for set_position_topic and get_position_topic

* check only set_position_topic in supported_features.

* Multidoc string fix.
This commit is contained in:
Pawel 2018-11-01 20:09:43 +01:00 • committed by Paulus Schoutsen
parent bcea3a9cba
commit c3e3f662f4
2 changed files with 369 additions and 116 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

162
homeassistant/components/cover/mqtt.py
View file

@ -34,9 +34,11 @@ _LOGGER = logging.getLogger(__name__)
DEPENDENCIES = ['mqtt']
CONF_GET_POSITION_TOPIC = 'position_topic'
CONF_TILT_COMMAND_TOPIC = 'tilt_command_topic'
CONF_TILT_STATUS_TOPIC = 'tilt_status_topic'
CONF_POSITION_TOPIC = 'set_position_topic'
CONF_SET_POSITION_TOPIC = 'set_position_topic'
CONF_SET_POSITION_TEMPLATE = 'set_position_template'
CONF_PAYLOAD_OPEN = 'payload_open'
@ -44,6 +46,8 @@ CONF_PAYLOAD_CLOSE = 'payload_close'
CONF_PAYLOAD_STOP = 'payload_stop'
CONF_STATE_OPEN = 'state_open'
CONF_STATE_CLOSED = 'state_closed'
CONF_POSITION_OPEN = 'position_open'
CONF_POSITION_CLOSED = 'position_closed'
CONF_TILT_CLOSED_POSITION = 'tilt_closed_value'
CONF_TILT_OPEN_POSITION = 'tilt_opened_value'
CONF_TILT_MIN = 'tilt_min'
@ -52,10 +56,15 @@ CONF_TILT_STATE_OPTIMISTIC = 'tilt_optimistic'
CONF_TILT_INVERT_STATE = 'tilt_invert_state'
CONF_UNIQUE_ID = 'unique_id'
TILT_PAYLOAD = "tilt"
COVER_PAYLOAD = "cover"
DEFAULT_NAME = 'MQTT Cover'
DEFAULT_PAYLOAD_OPEN = 'OPEN'
DEFAULT_PAYLOAD_CLOSE = 'CLOSE'
DEFAULT_PAYLOAD_STOP = 'STOP'
DEFAULT_POSITION_OPEN = 100
DEFAULT_POSITION_CLOSED = 0
DEFAULT_OPTIMISTIC = False
DEFAULT_RETAIN = False
DEFAULT_TILT_CLOSED_POSITION = 0
@ -69,11 +78,25 @@ OPEN_CLOSE_FEATURES = (SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_STOP)
TILT_FEATURES = (SUPPORT_OPEN_TILT | SUPPORT_CLOSE_TILT | SUPPORT_STOP_TILT |
SUPPORT_SET_TILT_POSITION)
PLATFORM_SCHEMA = mqtt.MQTT_BASE_PLATFORM_SCHEMA.extend({
def validate_options(value):
"""Validate options.
If set postion topic is set then get position topic is set as well.
"""
if (CONF_SET_POSITION_TOPIC in value and
CONF_GET_POSITION_TOPIC not in value):
raise vol.Invalid(
"Set position topic must be set together with get position topic.")
return value
PLATFORM_SCHEMA = vol.All(mqtt.MQTT_BASE_PLATFORM_SCHEMA.extend({
vol.Optional(CONF_COMMAND_TOPIC): valid_publish_topic,
vol.Optional(CONF_POSITION_TOPIC): valid_publish_topic,
vol.Optional(CONF_SET_POSITION_TOPIC): valid_publish_topic,
vol.Optional(CONF_SET_POSITION_TEMPLATE): cv.template,
vol.Optional(CONF_RETAIN, default=DEFAULT_RETAIN): cv.boolean,
vol.Optional(CONF_GET_POSITION_TOPIC): valid_subscribe_topic,
vol.Optional(CONF_STATE_TOPIC): valid_subscribe_topic,
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
@ -82,6 +105,10 @@ PLATFORM_SCHEMA = mqtt.MQTT_BASE_PLATFORM_SCHEMA.extend({
vol.Optional(CONF_PAYLOAD_STOP, default=DEFAULT_PAYLOAD_STOP): cv.string,
vol.Optional(CONF_STATE_OPEN, default=STATE_OPEN): cv.string,
vol.Optional(CONF_STATE_CLOSED, default=STATE_CLOSED): cv.string,
vol.Optional(CONF_POSITION_OPEN,
default=DEFAULT_POSITION_OPEN): int,
vol.Optional(CONF_POSITION_CLOSED,
default=DEFAULT_POSITION_CLOSED): int,
vol.Optional(CONF_OPTIMISTIC, default=DEFAULT_OPTIMISTIC): cv.boolean,
vol.Optional(CONF_TILT_COMMAND_TOPIC): valid_publish_topic,
vol.Optional(CONF_TILT_STATUS_TOPIC): valid_subscribe_topic,
@ -97,7 +124,7 @@ PLATFORM_SCHEMA = mqtt.MQTT_BASE_PLATFORM_SCHEMA.extend({
default=DEFAULT_TILT_INVERT_STATE): cv.boolean,
vol.Optional(CONF_UNIQUE_ID): cv.string,
vol.Optional(CONF_DEVICE): mqtt.MQTT_ENTITY_DEVICE_INFO_SCHEMA,
}).extend(mqtt.MQTT_AVAILABILITY_SCHEMA.schema)
}).extend(mqtt.MQTT_AVAILABILITY_SCHEMA.schema), validate_options)
async def async_setup_platform(hass: HomeAssistantType, config: ConfigType,
@ -132,6 +159,7 @@ async def _async_setup_entity(hass, config, async_add_entities,
async_add_entities([MqttCover(
config.get(CONF_NAME),
config.get(CONF_STATE_TOPIC),
config.get(CONF_GET_POSITION_TOPIC),
config.get(CONF_COMMAND_TOPIC),
config.get(CONF_AVAILABILITY_TOPIC),
config.get(CONF_TILT_COMMAND_TOPIC),
@ -140,6 +168,8 @@ async def _async_setup_entity(hass, config, async_add_entities,
config.get(CONF_RETAIN),
config.get(CONF_STATE_OPEN),
config.get(CONF_STATE_CLOSED),
config.get(CONF_POSITION_OPEN),
config.get(CONF_POSITION_CLOSED),
config.get(CONF_PAYLOAD_OPEN),
config.get(CONF_PAYLOAD_CLOSE),
config.get(CONF_PAYLOAD_STOP),
@ -153,7 +183,7 @@ async def _async_setup_entity(hass, config, async_add_entities,
config.get(CONF_TILT_MAX),
config.get(CONF_TILT_STATE_OPTIMISTIC),
config.get(CONF_TILT_INVERT_STATE),
config.get(CONF_POSITION_TOPIC),
config.get(CONF_SET_POSITION_TOPIC),
set_position_template,
config.get(CONF_UNIQUE_ID),
config.get(CONF_DEVICE),
@ -165,15 +195,16 @@ class MqttCover(MqttAvailability, MqttDiscoveryUpdate, MqttEntityDeviceInfo,
CoverDevice):
"""Representation of a cover that can be controlled using MQTT."""
def __init__(self, name, state_topic, command_topic, availability_topic,
def __init__(self, name, state_topic, get_position_topic,
command_topic, availability_topic,
tilt_command_topic, tilt_status_topic, qos, retain,
state_open, state_closed, payload_open, payload_close,
payload_stop, payload_available, payload_not_available,
optimistic, value_template, tilt_open_position,
tilt_closed_position, tilt_min, tilt_max, tilt_optimistic,
tilt_invert, position_topic, set_position_template,
unique_id: Optional[str], device_config: Optional[ConfigType],
discovery_hash):
state_open, state_closed, position_open, position_closed,
payload_open, payload_close, payload_stop, payload_available,
payload_not_available, optimistic, value_template,
tilt_open_position, tilt_closed_position, tilt_min, tilt_max,
tilt_optimistic, tilt_invert, set_position_topic,
set_position_template, unique_id: Optional[str],
device_config: Optional[ConfigType], discovery_hash):
"""Initialize the cover."""
MqttAvailability.__init__(self, availability_topic, qos,
payload_available, payload_not_available)
@ -183,6 +214,7 @@ class MqttCover(MqttAvailability, MqttDiscoveryUpdate, MqttEntityDeviceInfo,
self._state = None
self._name = name
self._state_topic = state_topic
self._get_position_topic = get_position_topic
self._command_topic = command_topic
self._tilt_command_topic = tilt_command_topic
self._tilt_status_topic = tilt_status_topic
@ -192,17 +224,20 @@ class MqttCover(MqttAvailability, MqttDiscoveryUpdate, MqttEntityDeviceInfo,
self._payload_stop = payload_stop
self._state_open = state_open
self._state_closed = state_closed
self._position_open = position_open
self._position_closed = position_closed
self._retain = retain
self._tilt_open_position = tilt_open_position
self._tilt_closed_position = tilt_closed_position
self._optimistic = optimistic or state_topic is None
self._optimistic = (optimistic or (state_topic is None and
get_position_topic is None))
self._template = value_template
self._tilt_value = None
self._tilt_min = tilt_min
self._tilt_max = tilt_max
self._tilt_optimistic = tilt_optimistic
self._tilt_invert = tilt_invert
self._position_topic = position_topic
self._set_position_topic = set_position_topic
self._set_position_template = set_position_template
self._unique_id = unique_id
self._discovery_hash = discovery_hash
@ -233,27 +268,43 @@ class MqttCover(MqttAvailability, MqttDiscoveryUpdate, MqttEntityDeviceInfo,
self._state = False
elif payload == self._state_closed:
self._state = True
elif payload.isnumeric() and 0 <= int(payload) <= 100:
if int(payload) > 0:
self._state = False
else:
self._state = True
self._position = int(payload)
else:
_LOGGER.warning(
"Payload is not True, False, or integer (0-100): %s",
payload)
_LOGGER.warning("Payload is not True or False: %s", payload)
return
self.async_schedule_update_ha_state()
if self._state_topic is None:
# Force into optimistic mode.
self._optimistic = True
else:
@callback
def position_message_received(topic, payload, qos):
"""Handle new MQTT state messages."""
if self._template is not None:
payload = self._template.async_render_with_possible_json_value(
payload)
if payload.isnumeric():
if 0 <= int(payload) <= 100:
percentage_payload = int(payload)
else:
percentage_payload = self.find_percentage_in_range(
float(payload), COVER_PAYLOAD)
if 0 <= percentage_payload <= 100:
self._position = percentage_payload
self._state = self._position == 0
else:
_LOGGER.warning(
"Payload is not integer within range: %s",
payload)
return
self.async_schedule_update_ha_state()
if self._get_position_topic:
await mqtt.async_subscribe(
self.hass, self._get_position_topic,
position_message_received, self._qos)
elif self._state_topic:
await mqtt.async_subscribe(
self.hass, self._state_topic,
state_message_received, self._qos)
else:
# Force into optimistic mode.
self._optimistic = True
if self._tilt_status_topic is None:
self._tilt_optimistic = True
@ -303,7 +354,7 @@ class MqttCover(MqttAvailability, MqttDiscoveryUpdate, MqttEntityDeviceInfo,
if self._command_topic is not None:
supported_features = OPEN_CLOSE_FEATURES
if self._position_topic is not None:
if self._set_position_topic is not None:
supported_features |= SUPPORT_SET_POSITION
if self._tilt_command_topic is not None:
@ -322,6 +373,8 @@ class MqttCover(MqttAvailability, MqttDiscoveryUpdate, MqttEntityDeviceInfo,
if self._optimistic:
# Optimistically assume that cover has changed state.
self._state = False
if self._get_position_topic:
self._position = self._position_open
self.async_schedule_update_ha_state()
async def async_close_cover(self, **kwargs):
@ -335,6 +388,8 @@ class MqttCover(MqttAvailability, MqttDiscoveryUpdate, MqttEntityDeviceInfo,
if self._optimistic:
# Optimistically assume that cover has changed state.
self._state = True
if self._get_position_topic:
self._position = self._position_closed
self.async_schedule_update_ha_state()
async def async_stop_cover(self, **kwargs):
@ -381,6 +436,7 @@ class MqttCover(MqttAvailability, MqttDiscoveryUpdate, MqttEntityDeviceInfo,
"""Move the cover to a specific position."""
if ATTR_POSITION in kwargs:
position = kwargs[ATTR_POSITION]
percentage_position = position
if self._set_position_template is not None:
try:
position = self._set_position_template.async_render(
@ -388,23 +444,36 @@ class MqttCover(MqttAvailability, MqttDiscoveryUpdate, MqttEntityDeviceInfo,
except TemplateError as ex:
_LOGGER.error(ex)
self._state = None
elif self._position_open != 100 and self._position_closed != 0:
position = self.find_in_range_from_percent(
position, COVER_PAYLOAD)
mqtt.async_publish(self.hass, self._position_topic,
mqtt.async_publish(self.hass, self._set_position_topic,
position, self._qos, self._retain)
if self._optimistic:
self._state = percentage_position == 0
self._position = percentage_position
self.async_schedule_update_ha_state()
def find_percentage_in_range(self, position):
def find_percentage_in_range(self, position, range_type=TILT_PAYLOAD):
"""Find the 0-100% value within the specified range."""
# the range of motion as defined by the min max values
tilt_range = self._tilt_max - self._tilt_min
if range_type == COVER_PAYLOAD:
max_range = self._position_open
min_range = self._position_closed
else:
max_range = self._tilt_max
min_range = self._tilt_min
current_range = max_range - min_range
# offset to be zero based
offset_position = position - self._tilt_min
# the percentage value within the range
position_percentage = float(offset_position) / tilt_range * 100.0
if self._tilt_invert:
offset_position = position - min_range
position_percentage = round(
float(offset_position) / current_range * 100.0)
if range_type == TILT_PAYLOAD and self._tilt_invert:
return 100 - position_percentage
return position_percentage
def find_in_range_from_percent(self, percentage):
def find_in_range_from_percent(self, percentage, range_type=TILT_PAYLOAD):
"""
Find the adjusted value for 0-100% within the specified range.
@ -413,14 +482,19 @@ class MqttCover(MqttAvailability, MqttDiscoveryUpdate, MqttEntityDeviceInfo,
by offsetting the max and min, getting the percentage value and
returning the offset
"""
offset = self._tilt_min
tilt_range = self._tilt_max - self._tilt_min
position = round(tilt_range * (percentage / 100.0))
if range_type == COVER_PAYLOAD:
max_range = self._position_open
min_range = self._position_closed
else:
max_range = self._tilt_max
min_range = self._tilt_min
offset = min_range
current_range = max_range - min_range
position = round(current_range * (percentage / 100.0))
position += offset
if self._tilt_invert:
position = self._tilt_max - position + offset
if range_type == TILT_PAYLOAD and self._tilt_invert:
position = max_range - position + offset
return position
@property

323
tests/components/cover/test_mqtt.py
View file

@ -50,24 +50,6 @@ class TestCoverMQTT(unittest.TestCase):
assert STATE_UNKNOWN == state.state
assert not state.attributes.get(ATTR_ASSUMED_STATE)
fire_mqtt_message(self.hass, 'state-topic', '0')
self.hass.block_till_done()
state = self.hass.states.get('cover.test')
assert STATE_CLOSED == state.state
fire_mqtt_message(self.hass, 'state-topic', '50')
self.hass.block_till_done()
state = self.hass.states.get('cover.test')
assert STATE_OPEN == state.state
fire_mqtt_message(self.hass, 'state-topic', '100')
self.hass.block_till_done()
state = self.hass.states.get('cover.test')
assert STATE_OPEN == state.state
fire_mqtt_message(self.hass, 'state-topic', STATE_CLOSED)
self.hass.block_till_done()
@ -80,6 +62,39 @@ class TestCoverMQTT(unittest.TestCase):
state = self.hass.states.get('cover.test')
assert STATE_OPEN == state.state
def test_position_via_position_topic(self):
"""Test the controlling state via topic."""
self.assertTrue(setup_component(self.hass, cover.DOMAIN, {
cover.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'position_topic': 'get-position-topic',
'position_open': 100,
'position_closed': 0,
'command_topic': 'command-topic',
'qos': 0,
'payload_open': 'OPEN',
'payload_close': 'CLOSE',
'payload_stop': 'STOP'
}
}))
state = self.hass.states.get('cover.test')
assert STATE_UNKNOWN == state.state
assert not state.attributes.get(ATTR_ASSUMED_STATE)
fire_mqtt_message(self.hass, 'get-position-topic', '0')
self.hass.block_till_done()
state = self.hass.states.get('cover.test')
assert STATE_CLOSED == state.state
fire_mqtt_message(self.hass, 'get-position-topic', '100')
self.hass.block_till_done()
state = self.hass.states.get('cover.test')
assert STATE_OPEN == state.state
def test_state_via_template(self):
"""Test the controlling state via topic."""
assert setup_component(self.hass, cover.DOMAIN, {
@ -89,7 +104,12 @@ class TestCoverMQTT(unittest.TestCase):
'state_topic': 'state-topic',
'command_topic': 'command-topic',
'qos': 0,
'value_template': '{{ (value | multiply(0.01)) | int }}',
'value_template': '\
{% if (value | multiply(0.01) | int) == 0 %}\
closed\
{% else %}\
open\
{% endif %}'
}
})
@ -108,6 +128,40 @@ class TestCoverMQTT(unittest.TestCase):
state = self.hass.states.get('cover.test')
assert STATE_CLOSED == state.state
def test_position_via_template(self):
"""Test the controlling state via topic."""
self.assertTrue(setup_component(self.hass, cover.DOMAIN, {
cover.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'position_topic': 'get-position-topic',
'command_topic': 'command-topic',
'qos': 0,
'value_template': '{{ (value | multiply(0.01)) | int }}'
}
}))
state = self.hass.states.get('cover.test')
self.assertEqual(STATE_UNKNOWN, state.state)
fire_mqtt_message(self.hass, 'get-position-topic', '10000')
self.hass.block_till_done()
state = self.hass.states.get('cover.test')
self.assertEqual(STATE_OPEN, state.state)
fire_mqtt_message(self.hass, 'get-position-topic', '5000')
self.hass.block_till_done()
state = self.hass.states.get('cover.test')
self.assertEqual(STATE_OPEN, state.state)
fire_mqtt_message(self.hass, 'get-position-topic', '99')
self.hass.block_till_done()
state = self.hass.states.get('cover.test')
self.assertEqual(STATE_CLOSED, state.state)
def test_optimistic_state_change(self):
"""Test changing state optimistically."""
assert setup_component(self.hass, cover.DOMAIN, {
@ -225,8 +279,10 @@ class TestCoverMQTT(unittest.TestCase):
cover.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'position_topic': 'get-position-topic',
'command_topic': 'command-topic',
'position_open': 100,
'position_closed': 0,
'payload_open': 'OPEN',
'payload_close': 'CLOSE',
'payload_stop': 'STOP'
@ -240,25 +296,25 @@ class TestCoverMQTT(unittest.TestCase):
assert not (4 & self.hass.states.get(
'cover.test').attributes['supported_features'] == 4)
fire_mqtt_message(self.hass, 'state-topic', '0')
fire_mqtt_message(self.hass, 'get-position-topic', '0')
self.hass.block_till_done()
current_cover_position = self.hass.states.get(
'cover.test').attributes['current_position']
assert 0 == current_cover_position
fire_mqtt_message(self.hass, 'state-topic', '50')
fire_mqtt_message(self.hass, 'get-position-topic', '50')
self.hass.block_till_done()
current_cover_position = self.hass.states.get(
'cover.test').attributes['current_position']
assert 50 == current_cover_position
fire_mqtt_message(self.hass, 'state-topic', '101')
fire_mqtt_message(self.hass, 'get-position-topic', '101')
self.hass.block_till_done()
current_cover_position = self.hass.states.get(
'cover.test').attributes['current_position']
assert 50 == current_cover_position
fire_mqtt_message(self.hass, 'state-topic', 'non-numeric')
fire_mqtt_message(self.hass, 'get-position-topic', 'non-numeric')
self.hass.block_till_done()
current_cover_position = self.hass.states.get(
'cover.test').attributes['current_position']
@ -270,9 +326,11 @@ class TestCoverMQTT(unittest.TestCase):
cover.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'position_topic': 'get-position-topic',
'command_topic': 'command-topic',
'set_position_topic': 'position-topic',
'set_position_topic': 'set-position-topic',
'position_open': 100,
'position_closed': 0,
'payload_open': 'OPEN',
'payload_close': 'CLOSE',
'payload_stop': 'STOP'
@ -283,11 +341,10 @@ class TestCoverMQTT(unittest.TestCase):
'cover.test').attributes
assert not ('current_position' in state_attributes_dict)
assert not ('current_tilt_position' in state_attributes_dict)
assert 4 & self.hass.states.get(
'cover.test').attributes['supported_features'] == 4
fire_mqtt_message(self.hass, 'state-topic', '22')
fire_mqtt_message(self.hass, 'get-position-topic', '22')
self.hass.block_till_done()
state_attributes_dict = self.hass.states.get(
'cover.test').attributes
@ -303,9 +360,11 @@ class TestCoverMQTT(unittest.TestCase):
cover.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'position_topic': 'get-position-topic',
'command_topic': 'command-topic',
'set_position_topic': 'position-topic',
'position_open': 100,
'position_closed': 0,
'set_position_topic': 'set-position-topic',
'set_position_template': '{{100-62}}',
'payload_open': 'OPEN',
'payload_close': 'CLOSE',
@ -319,7 +378,7 @@ class TestCoverMQTT(unittest.TestCase):
self.hass.block_till_done()
self.mock_publish.async_publish.assert_called_once_with(
'position-topic', '38', 0, False)
'set-position-topic', '38', 0, False)
def test_set_position_untemplated(self):
"""Test setting cover position via template."""
@ -327,7 +386,7 @@ class TestCoverMQTT(unittest.TestCase):
cover.DOMAIN: {
'platform': 'mqtt',
'name': 'test',
'state_topic': 'state-topic',
'position_topic': 'state-topic',
'command_topic': 'command-topic',
'set_position_topic': 'position-topic',
'payload_open': 'OPEN',
@ -612,90 +671,210 @@ class TestCoverMQTT(unittest.TestCase):
def test_find_percentage_in_range_defaults(self):
"""Test find percentage in range with default range."""
mqtt_cover = MqttCover(
'cover.test', 'state-topic', 'command-topic', None,
'tilt-command-topic', 'tilt-status-topic', 0, False,
'OPEN', 'CLOSE', 'OPEN', 'CLOSE', 'STOP', None, None,
False, None, 100, 0, 0, 100, False, False, None, None, None,
None, None)
name='cover.test',
state_topic='state-topic',
get_position_topic=None,
command_topic='command-topic',
availability_topic=None,
tilt_command_topic='tilt-command-topic',
tilt_status_topic='tilt-status-topic',
qos=0,
retain=False,
state_open='OPEN', state_closed='CLOSE',
position_open=100, position_closed=0,
payload_open='OPEN', payload_close='CLOSE', payload_stop='STOP',
payload_available=None, payload_not_available=None,
optimistic=False, value_template=None,
tilt_open_position=100, tilt_closed_position=0,
tilt_min=0, tilt_max=100, tilt_optimistic=False,
tilt_invert=False,
set_position_topic=None, set_position_template=None,
unique_id=None, device_config=None, discovery_hash=None)
assert 44 == mqtt_cover.find_percentage_in_range(44)
assert 44 == mqtt_cover.find_percentage_in_range(44, 'cover')
def test_find_percentage_in_range_altered(self):
"""Test find percentage in range with altered range."""
mqtt_cover = MqttCover(
'cover.test', 'state-topic', 'command-topic', None,
'tilt-command-topic', 'tilt-status-topic', 0, False,
'OPEN', 'CLOSE', 'OPEN', 'CLOSE', 'STOP', None, None,
False, None, 180, 80, 80, 180, False, False, None, None, None,
None, None)
name='cover.test',
state_topic='state-topic',
get_position_topic=None,
command_topic='command-topic',
availability_topic=None,
tilt_command_topic='tilt-command-topic',
tilt_status_topic='tilt-status-topic',
qos=0,
retain=False,
state_open='OPEN', state_closed='CLOSE',
position_open=180, position_closed=80,
payload_open='OPEN', payload_close='CLOSE', payload_stop='STOP',
payload_available=None, payload_not_available=None,
optimistic=False, value_template=None,
tilt_open_position=180, tilt_closed_position=80,
tilt_min=80, tilt_max=180, tilt_optimistic=False,
tilt_invert=False,
set_position_topic=None, set_position_template=None,
unique_id=None, device_config=None, discovery_hash=None)
assert 40 == mqtt_cover.find_percentage_in_range(120)
assert 40 == mqtt_cover.find_percentage_in_range(120, 'cover')
def test_find_percentage_in_range_defaults_inverted(self):
"""Test find percentage in range with default range but inverted."""
mqtt_cover = MqttCover(
'cover.test', 'state-topic', 'command-topic', None,
'tilt-command-topic', 'tilt-status-topic', 0, False,
'OPEN', 'CLOSE', 'OPEN', 'CLOSE', 'STOP', None, None,
False, None, 100, 0, 0, 100, False, True, None, None, None,
None, None)
name='cover.test',
state_topic='state-topic',
get_position_topic=None,
command_topic='command-topic',
availability_topic=None,
tilt_command_topic='tilt-command-topic',
tilt_status_topic='tilt-status-topic',
qos=0,
retain=False,
state_open='OPEN', state_closed='CLOSE',
position_open=0, position_closed=100,
payload_open='OPEN', payload_close='CLOSE', payload_stop='STOP',
payload_available=None, payload_not_available=None,
optimistic=False, value_template=None,
tilt_open_position=100, tilt_closed_position=0,
tilt_min=0, tilt_max=100, tilt_optimistic=False,
tilt_invert=True,
set_position_topic=None, set_position_template=None,
unique_id=None, device_config=None, discovery_hash=None)
assert 56 == mqtt_cover.find_percentage_in_range(44)
assert 56 == mqtt_cover.find_percentage_in_range(44, 'cover')
def test_find_percentage_in_range_altered_inverted(self):
"""Test find percentage in range with altered range and inverted."""
mqtt_cover = MqttCover(
'cover.test', 'state-topic', 'command-topic', None,
'tilt-command-topic', 'tilt-status-topic', 0, False,
'OPEN', 'CLOSE', 'OPEN', 'CLOSE', 'STOP', None, None,
False, None, 180, 80, 80, 180, False, True, None, None, None,
None, None)
name='cover.test',
state_topic='state-topic',
get_position_topic=None,
command_topic='command-topic',
availability_topic=None,
tilt_command_topic='tilt-command-topic',
tilt_status_topic='tilt-status-topic',
qos=0,
retain=False,
state_open='OPEN', state_closed='CLOSE',
position_open=80, position_closed=180,
payload_open='OPEN', payload_close='CLOSE', payload_stop='STOP',
payload_available=None, payload_not_available=None,
optimistic=False, value_template=None,
tilt_open_position=180, tilt_closed_position=80,
tilt_min=80, tilt_max=180, tilt_optimistic=False,
tilt_invert=True,
set_position_topic=None, set_position_template=None,
unique_id=None, device_config=None, discovery_hash=None)
assert 60 == mqtt_cover.find_percentage_in_range(120)
assert 60 == mqtt_cover.find_percentage_in_range(120, 'cover')
def test_find_in_range_defaults(self):
"""Test find in range with default range."""
mqtt_cover = MqttCover(
'cover.test', 'state-topic', 'command-topic', None,
'tilt-command-topic', 'tilt-status-topic', 0, False,
'OPEN', 'CLOSE', 'OPEN', 'CLOSE', 'STOP', None, None,
False, None, 100, 0, 0, 100, False, False, None, None, None,
None, None)
name='cover.test',
state_topic='state-topic',
get_position_topic=None,
command_topic='command-topic',
availability_topic=None,
tilt_command_topic='tilt-command-topic',
tilt_status_topic='tilt-status-topic',
qos=0,
retain=False,
state_open='OPEN', state_closed='CLOSE',
position_open=100, position_closed=0,
payload_open='OPEN', payload_close='CLOSE', payload_stop='STOP',
payload_available=None, payload_not_available=None,
optimistic=False, value_template=None,
tilt_open_position=100, tilt_closed_position=0,
tilt_min=0, tilt_max=100, tilt_optimistic=False,
tilt_invert=False,
set_position_topic=None, set_position_template=None,
unique_id=None, device_config=None, discovery_hash=None)
assert 44 == mqtt_cover.find_in_range_from_percent(44)
assert 44 == mqtt_cover.find_in_range_from_percent(44, 'cover')
def test_find_in_range_altered(self):
"""Test find in range with altered range."""
mqtt_cover = MqttCover(
'cover.test', 'state-topic', 'command-topic', None,
'tilt-command-topic', 'tilt-status-topic', 0, False,
'OPEN', 'CLOSE', 'OPEN', 'CLOSE', 'STOP', None, None,
False, None, 180, 80, 80, 180, False, False, None, None, None,
None, None)
name='cover.test',
state_topic='state-topic',
get_position_topic=None,
command_topic='command-topic',
availability_topic=None,
tilt_command_topic='tilt-command-topic',
tilt_status_topic='tilt-status-topic',
qos=0,
retain=False,
state_open='OPEN', state_closed='CLOSE',
position_open=180, position_closed=80,
payload_open='OPEN', payload_close='CLOSE', payload_stop='STOP',
payload_available=None, payload_not_available=None,
optimistic=False, value_template=None,
tilt_open_position=180, tilt_closed_position=80,
tilt_min=80, tilt_max=180, tilt_optimistic=False,
tilt_invert=False,
set_position_topic=None, set_position_template=None,
unique_id=None, device_config=None, discovery_hash=None)
assert 120 == mqtt_cover.find_in_range_from_percent(40)
assert 120 == mqtt_cover.find_in_range_from_percent(40, 'cover')
def test_find_in_range_defaults_inverted(self):
"""Test find in range with default range but inverted."""
mqtt_cover = MqttCover(
'cover.test', 'state-topic', 'command-topic', None,
'tilt-command-topic', 'tilt-status-topic', 0, False,
'OPEN', 'CLOSE', 'OPEN', 'CLOSE', 'STOP', None, None,
False, None, 100, 0, 0, 100, False, True, None, None, None,
None, None)
name='cover.test',
state_topic='state-topic',
get_position_topic=None,
command_topic='command-topic',
availability_topic=None,
tilt_command_topic='tilt-command-topic',
tilt_status_topic='tilt-status-topic',
qos=0,
retain=False,
state_open='OPEN', state_closed='CLOSE',
position_open=0, position_closed=100,
payload_open='OPEN', payload_close='CLOSE', payload_stop='STOP',
payload_available=None, payload_not_available=None,
optimistic=False, value_template=None,
tilt_open_position=100, tilt_closed_position=0,
tilt_min=0, tilt_max=100, tilt_optimistic=False,
tilt_invert=True,
set_position_topic=None, set_position_template=None,
unique_id=None, device_config=None, discovery_hash=None)
assert 44 == mqtt_cover.find_in_range_from_percent(56)
assert 44 == mqtt_cover.find_in_range_from_percent(56, 'cover')
def test_find_in_range_altered_inverted(self):
"""Test find in range with altered range and inverted."""
mqtt_cover = MqttCover(
'cover.test', 'state-topic', 'command-topic', None,
'tilt-command-topic', 'tilt-status-topic', 0, False,
'OPEN', 'CLOSE', 'OPEN', 'CLOSE', 'STOP', None, None,
False, None, 180, 80, 80, 180, False, True, None, None, None,
None, None)
name='cover.test',
state_topic='state-topic',
get_position_topic=None,
command_topic='command-topic',
availability_topic=None,
tilt_command_topic='tilt-command-topic',
tilt_status_topic='tilt-status-topic',
qos=0,
retain=False,
state_open='OPEN', state_closed='CLOSE',
position_open=80, position_closed=180,
payload_open='OPEN', payload_close='CLOSE', payload_stop='STOP',
payload_available=None, payload_not_available=None,
optimistic=False, value_template=None,
tilt_open_position=180, tilt_closed_position=80,
tilt_min=80, tilt_max=180, tilt_optimistic=False,
tilt_invert=True,
set_position_topic=None, set_position_template=None,
unique_id=None, device_config=None, discovery_hash=None)
assert 120 == mqtt_cover.find_in_range_from_percent(60)
assert 120 == mqtt_cover.find_in_range_from_percent(60, 'cover')
def test_availability_without_topic(self):
"""Test availability without defined availability topic."""