"""Switcher Bridge Response Messages."""
# fmt: off
from asyncio import AbstractEventLoop, Future, ensure_future
from binascii import hexlify
from socket import inet_ntoa
from struct import pack
from typing import Optional, Union
from ..consts import (ENCODING_CODEC, STATE_OFF, STATE_ON, STATE_RESPONSE_ON,
WAITING_TEXT)
from ..tools import convert_seconds_to_iso_time
# fmt: on
[docs]class SwitcherV2BroadcastMSG:
"""Represntation of the SwitcherV2 broadcast message.
Args:
loop: the event loop to perform actions in.
message: the raw message from the device.
Todo:
* replace ``init_future`` attribute with ``get_init_future`` method.
"""
def __init__(
self, loop: AbstractEventLoop, message: Union[bytes, str]
) -> None:
"""Initialize the broadcast message."""
self._loop = loop
self._verified = self._validated = False
self._power_consumption = 0
self._electric_current = 0.0
self._ip_address = WAITING_TEXT
self._mac_address = WAITING_TEXT
self._name = WAITING_TEXT
self._device_id = WAITING_TEXT
self._auto_off_set = WAITING_TEXT
self._remaining_time_to_off = WAITING_TEXT
self._init_future = loop.create_future()
fixed_msg = (
message
if isinstance(message, bytes)
else message.encode(ENCODING_CODEC)
)
ensure_future(self.initialize(fixed_msg), loop=loop)
[docs] async def initialize(self, message: bytes) -> None:
"""Finish the initialization and update the future object.
Args:
message: the raw message from the device.
"""
try:
self._verified = (
hexlify(message)[0:4].decode(ENCODING_CODEC) == "fef0"
and len(message) == 165
)
if self._verified:
temp_ip = hexlify(message)[152:160]
ip_addr = int(
temp_ip[6:8] + temp_ip[4:6] + temp_ip[2:4] + temp_ip[0:2],
16,
)
self._ip_address = inet_ntoa(pack("<L", ip_addr))
mac = hexlify(message)[160:172].decode(ENCODING_CODEC).upper()
self._mac_address = (
mac[0:2]
+ ":"
+ mac[2:4]
+ ":"
+ mac[4:6]
+ ":"
+ mac[6:8]
+ ":"
+ mac[8:10]
+ ":"
+ mac[10:12]
)
self._name = (
message[42:74].decode(ENCODING_CODEC).rstrip("\x00")
)
self._device_id = hexlify(message)[36:42].decode(
ENCODING_CODEC
)
self._device_state = (
STATE_ON
if hexlify(message)[266:270].decode(ENCODING_CODEC)
== STATE_RESPONSE_ON
else STATE_OFF
)
temp_auto_off_set = hexlify(message)[310:318]
temp_auto_off_set_seconds = int(
temp_auto_off_set[6:8]
+ temp_auto_off_set[4:6]
+ temp_auto_off_set[2:4]
+ temp_auto_off_set[0:2],
16,
)
self._auto_off_set = await convert_seconds_to_iso_time(
self._loop, temp_auto_off_set_seconds
)
if self._device_state == STATE_ON:
temp_power = hexlify(message)[270:278]
self._power_consumption = int(
temp_power[2:4] + temp_power[0:2], 16
)
self._electric_current = round(
(self._power_consumption / float(220)), 1
)
temp_remaining_time = hexlify(message)[294:302]
temp_remaining_time_seconds = int(
temp_remaining_time[6:8]
+ temp_remaining_time[4:6]
+ temp_remaining_time[2:4]
+ temp_remaining_time[0:2],
16,
)
# TODO: black keeps pushing this lengthy line, why?
self._remaining_time_to_off = await convert_seconds_to_iso_time( # noqa E501
self._loop, temp_remaining_time_seconds
)
self._validated = True
self.init_future.set_result(self)
except (ValueError, IndexError, RuntimeError) as ex:
self.init_future.set_exception(ex)
return None
@property
def verified(self) -> bool:
"""bool: Return whether or not the message is a SwitcherV2 message."""
return self._verified if self._validated else self._validated
@property
def ip_address(self) -> str:
"""str: Return the ip address."""
return self._ip_address
@property
def mac_address(self) -> str:
"""str: Return the mac address."""
return self._mac_address
@property
def name(self) -> str:
"""str: Return the device name."""
return self._name
@property
def device_id(self) -> str:
"""str: Return the device id."""
return self._device_id
@property
def power(self) -> int:
"""int: Return the power consumptionin watts."""
return self._power_consumption
@property
def device_state(self) -> str:
"""str: Return the state of the device."""
return self._device_state
@property
def remaining_time_to_off(self) -> Optional[str]:
"""str: Return the time left to auto-off."""
return self._remaining_time_to_off
@property
def current(self) -> float:
"""float: Return the power consumptionin amps."""
return self._electric_current
@property
def auto_off_set(self) -> str:
"""str: Return the auto-off configuration value."""
return self._auto_off_set
@property
def init_future(self) -> Future:
"""asyncio.Future: Return the future of the device initialization."""
return self._init_future