diff --git a/custom_components/mlgw/__init__.py b/custom_components/mlgw/__init__.py
index 5b542d9..b6aecfe 100644
--- a/custom_components/mlgw/__init__.py
+++ b/custom_components/mlgw/__init__.py
@@ -1,49 +1,55 @@
"""The MasterLink Gateway integration."""
+
import asyncio
-from homeassistant.helpers import device_registry as dr
-from homeassistant.helpers.typing import ServiceDataType
-from .gateway import create_mlgw_gateway, MasterLinkGateway
-from .const import reverse_ml_destselectordict, reverse_ml_selectedsourcedict, BEO4_CMDS
-from .media_player import BeoSpeaker
import logging
-import json
+import requests
+from requests.auth import HTTPBasicAuth, HTTPDigestAuth
+from requests.exceptions import RequestException
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigEntryNotReady
from homeassistant.const import (
+ CONF_DEVICES,
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
- CONF_DEVICES,
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
-from homeassistant.helpers import config_validation as cv, discovery
+from homeassistant.helpers import (
+ config_validation as cv,
+ device_registry as dr,
+ discovery,
+)
from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.typing import ServiceDataType
from .const import (
- DOMAIN,
- MLGW_GATEWAY,
- MLGW_GATEWAY_CONFIGURATION_DATA,
- MLGW_DEVICES,
- CONF_MLGW_DEFAULT_SOURCE,
+ ATTR_MLGW_ACTION,
+ ATTR_MLGW_BUTTON,
+ BASE_URL,
+ BEO4_CMDS,
CONF_MLGW_AVAILABLE_SOURCES,
- MLGW_DEFAULT_SOURCE,
- MLGW_AVAILABLE_SOURCES,
- CONF_MLGW_DEVICE_NAME,
+ CONF_MLGW_DEFAULT_SOURCE,
+ CONF_MLGW_DEVICE_MLID,
CONF_MLGW_DEVICE_MLN,
+ CONF_MLGW_DEVICE_NAME,
CONF_MLGW_DEVICE_ROOM,
- CONF_MLGW_DEVICE_MLID,
CONF_MLGW_USE_MLLOG,
- BASE_URL,
- TIMEOUT,
+ DOMAIN,
+ MLGW_AVAILABLE_SOURCES,
MLGW_CONFIG_JSON_PATH,
- ATTR_MLGW_ACTION,
- ATTR_MLGW_BUTTON,
+ MLGW_DEFAULT_SOURCE,
+ MLGW_DEVICES,
+ MLGW_GATEWAY,
+ MLGW_GATEWAY_CONFIGURATION_DATA,
+ TIMEOUT,
+ reverse_ml_destselectordict,
+ reverse_ml_selectedsourcedict,
reverse_mlgw_virtualactiondict,
)
-
+from .gateway import MasterLinkGateway, create_mlgw_gateway
CONFIG_SCHEMA = vol.Schema(
{
@@ -97,8 +103,8 @@
def yaml_to_json_config(manual_devices, availabe_sources):
"""Convert the YAML configuration into the equivalent json config from the MLGW."""
- result = dict()
- result["zones"] = list()
+ result = {}
+ result["zones"] = []
i = 1
for device in manual_devices:
if CONF_MLGW_DEVICE_MLN in device.keys():
@@ -116,7 +122,7 @@ def yaml_to_json_config(manual_devices, availabe_sources):
r = 0
_zl = [w for w in result["zones"] if w["number"] == room]
if len(_zl) == 0:
- _z = dict()
+ _z = {}
_z["number"] = room
result["zones"].append(_z)
r = result["zones"].index(_z)
@@ -124,16 +130,16 @@ def yaml_to_json_config(manual_devices, availabe_sources):
r = result["zones"].index(_zl[0])
if "products" not in result["zones"][r]:
- result["zones"][r]["products"] = list()
+ result["zones"][r]["products"] = []
- product = dict()
+ product = {}
product["MLN"] = mln
product["ML"] = ml
product["name"] = device[CONF_MLGW_DEVICE_NAME]
- device_sources = list()
+ device_sources = []
for _x in availabe_sources:
- _source = dict()
+ _source = {}
_source["name"] = _x
_source["destination"] = reverse_ml_destselectordict.get("AUDIO SOURCE")
_source["format"] = "F0"
@@ -141,7 +147,7 @@ def yaml_to_json_config(manual_devices, availabe_sources):
_source["link"] = 0
_source["statusID"] = reverse_ml_selectedsourcedict.get(_x)
_source["selectID"] = BEO4_CMDS.get(_x)
- _source["selectCmds"] = list()
+ _source["selectCmds"] = []
_source["selectCmds"].append({"cmd": BEO4_CMDS.get(_x), "format": "F0"})
device_sources.append(_source)
@@ -207,8 +213,6 @@ async def async_setup(hass: HomeAssistant, config: dict):
def get_mlgw_configuration_data(host: str, username: str, password: str):
"""Get the configuration data from the mlgw using the mlgwpservices.json endpoint."""
- import requests
- from requests.auth import HTTPDigestAuth, HTTPBasicAuth
# try with Digest Auth first
response = requests.get(
@@ -245,6 +249,7 @@ def send_all_standby(service: ServiceDataType):
if not gateway:
return False
gateway.mlgw_send_all_standby()
+ return True
# Register the services
hass.services.async_register(
@@ -264,7 +269,6 @@ def send_all_standby(service: ServiceDataType):
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up MasterLink Gateway from a config entry."""
- from requests.exceptions import RequestException
host = entry.data.get(CONF_HOST)
password = entry.data.get(CONF_PASSWORD)
@@ -275,7 +279,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
mlgw_configurationdata = await hass.async_add_executor_job(
get_mlgw_configuration_data, host, username, password
)
- except (RequestException) as ex:
+ except RequestException as ex:
# this will cause Home Assistant to retry setting up the integration later.
raise ConfigEntryNotReady(f"Cannot connect to {host}, is it on?") from ex
@@ -296,9 +300,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
hass.data[DOMAIN][entry.entry_id] = {}
hass.data[DOMAIN][entry.entry_id][MLGW_GATEWAY] = gateway
- hass.data[DOMAIN][entry.entry_id][
- MLGW_GATEWAY_CONFIGURATION_DATA
- ] = mlgw_configurationdata
+ hass.data[DOMAIN][entry.entry_id][MLGW_GATEWAY_CONFIGURATION_DATA] = (
+ mlgw_configurationdata
+ )
hass.data[DOMAIN][entry.entry_id]["serial"] = entry.unique_id
_LOGGER.debug("Serial: %s", entry.unique_id)
@@ -314,10 +318,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(**device_info)
- for component in PLATFORMS:
- hass.async_create_task(
- hass.config_entries.async_forward_entry_setup(entry, component)
- )
+ # for component in PLATFORMS:
+ # hass.async_create_task(
+ # hass.config_entries.async_forward_entry_setup(entry, component)
+ # )
+ await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
register_services(hass, gateway)
diff --git a/custom_components/mlgw/config_flow.py b/custom_components/mlgw/config_flow.py
index b0ddfc5..4a8522e 100644
--- a/custom_components/mlgw/config_flow.py
+++ b/custom_components/mlgw/config_flow.py
@@ -1,34 +1,33 @@
-"""
-
-Config flow for MasterLink Gateway integration.
+"""Config flow for MasterLink Gateway integration.
Includes code from Lele-72. Thank you!
"""
-import logging
+
import ipaddress
+import logging
import re
-import voluptuous as vol
-import requests
import socket
import xml.etree.ElementTree as ET
-from requests.auth import HTTPDigestAuth, HTTPBasicAuth
+
+import requests
+from requests.auth import HTTPBasicAuth, HTTPDigestAuth
from requests.exceptions import ConnectTimeout
+import voluptuous as vol
-from homeassistant import config_entries, core, exceptions, data_entry_flow
+from homeassistant import config_entries, core, data_entry_flow, exceptions
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from .const import (
- CONF_MLGW_USE_MLLOG,
BASE_URL,
+ CONF_MLGW_USE_MLLOG,
+ DOMAIN,
MLGW_CONFIG_JSON_PATH,
TIMEOUT,
) # pylint:disable=unused-import
-from .const import DOMAIN
-
_LOGGER = logging.getLogger(__name__)
-_Discovered_MLGW : dict = { }
+_Discovered_MLGW: dict = {}
# Data schema for the configuration flows
USER_STEP_DATA_SCHEMA = vol.Schema(
@@ -60,6 +59,7 @@ def host_valid(host):
## Get serial number of mlgw
+
# Get the MLGW/BLGW Serial number from the integrated Jabber Client
#
# This code causes the MLGW to crash if called too many times. Given that it is called every time there is
@@ -72,8 +72,8 @@ def host_valid(host):
#
#
-async def mlgw_get_xmpp_serial(_host: str) -> str:
+async def mlgw_get_xmpp_serial(_host: str) -> str:
if _host in _Discovered_MLGW:
_LOGGER.debug("XMPP found cached sn")
return _Discovered_MLGW[_host]
@@ -117,7 +117,7 @@ async def mlgw_get_xmpp_serial(_host: str) -> str:
class CheckPasswordMLGWHub:
"""Checks Password for the MLGW Hub and gets basic information."""
- def __init__(self, host):
+ def __init__(self, host) -> None:
"""Initialize."""
self._host = host
self._data = None
@@ -140,7 +140,7 @@ def authenticate(self, user, password) -> bool:
if response.status_code == 401:
_LOGGER.warning("Invalid authentication to MLGW")
- raise InvalidAuth()
+ raise InvalidAuth
if response.status_code != 200:
return False
@@ -184,7 +184,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
# TODO pick one of the available connection classes in homeassistant/config_entries.py
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
- def __init__(self):
+ def __init__(self) -> None:
"""Initialize."""
self.host = None
@@ -201,7 +201,7 @@ async def async_step_user(self, user_input=None):
try:
if not host_valid(user_input[CONF_HOST]):
- raise InvalidHost()
+ raise InvalidHost
info = await validate_input(self.hass, user_input)
except CannotConnect:
@@ -284,7 +284,7 @@ async def async_step_zeroconf_confirm(self, user_input=None):
try:
if not host_valid(user_input[CONF_HOST]):
- raise InvalidHost()
+ raise InvalidHost
info = await validate_input(self.hass, user_input)
diff --git a/custom_components/mlgw/device_action.py b/custom_components/mlgw/device_action.py
index 23bc135..0ac4365 100644
--- a/custom_components/mlgw/device_action.py
+++ b/custom_components/mlgw/device_action.py
@@ -1,20 +1,22 @@
-"""
+"""Provides Device Actions for the integration.
+
Provides the following Device Actions:
- Virtual Buttons
- All Standby
+Virtual Buttons
+All Standby
"""
+
from __future__ import annotations
-import homeassistant.helpers.config_validation as cv
import voluptuous as vol
+
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_TYPE
from homeassistant.core import Context, HomeAssistant
-from homeassistant.helpers import entity_registry
+from homeassistant.helpers import entity_registry as er
+import homeassistant.helpers.config_validation as cv
+from . import DOMAIN, SERVICE_ALL_STANDBY, SERVICE_VIRTUAL_BUTTON
from .const import ATTR_MLGW_BUTTON
-from . import DOMAIN, SERVICE_VIRTUAL_BUTTON, SERVICE_ALL_STANDBY
-
ACTION_TYPES = {"all_standby", "virtual_button"}
ALL_STANDBY_SCHEMA = cv.DEVICE_ACTION_BASE_SCHEMA.extend(
@@ -41,10 +43,10 @@ async def async_get_actions(
hass: HomeAssistant, device_id: str
) -> list[dict[str, str]]:
"""List device actions for aketest_integration devices."""
- registry = entity_registry.async_get(hass)
+ registry = er.async_get(hass)
actions = []
- if not entity_registry.async_entries_for_device(registry, device_id):
+ if not er.async_entries_for_device(registry, device_id):
base_action = {
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
diff --git a/custom_components/mlgw/device_trigger.py b/custom_components/mlgw/device_trigger.py
index fd5c3e9..5e8ab43 100644
--- a/custom_components/mlgw/device_trigger.py
+++ b/custom_components/mlgw/device_trigger.py
@@ -1,36 +1,33 @@
-"""
- Provides device triggers for the MasterLink Gateway integration.
+"""Provides device triggers for the MasterLink Gateway integration.
- Triggers include:
- * Control
- * Light
- for commonly used keys on the beo4 remote.
+Triggers include:
+* Control
+* Light
+for commonly used keys on the beo4 remote.
- Triggers will also include the room name or number (if no name is available).
+Triggers will also include the room name or number (if no name is available).
"""
+
from __future__ import annotations
from typing import Any
import voluptuous as vol
-from homeassistant.helpers.trigger import (
- TriggerActionType,
- TriggerInfo,
-)
+
from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA
from homeassistant.components.homeassistant.triggers import event as event_trigger
from homeassistant.const import (
CONF_DEVICE_ID,
CONF_DOMAIN,
CONF_PLATFORM,
- CONF_TYPE,
CONF_ROOM,
+ CONF_TYPE,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
-from homeassistant.helpers import entity_registry
+from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType
-from homeassistant.helpers import config_validation as cv
from .const import DOMAIN, MLGW_EVENT_MLGW_TELEGRAM
@@ -143,10 +140,10 @@ async def async_get_triggers(
) -> list[dict[str, Any]]:
"""List device triggers for mlgw integration devices."""
- registry = entity_registry.async_get(hass)
+ registry = er.async_get(hass)
triggers = []
- if not entity_registry.async_entries_for_device(registry, device_id):
+ if not er.async_entries_for_device(registry, device_id):
base_trigger = {
CONF_PLATFORM: "device",
CONF_DEVICE_ID: device_id,
@@ -154,16 +151,16 @@ async def async_get_triggers(
ROOM_ID: "",
}
- for triggertype in TRIGGER_TYPES:
- for light_command in LIGHT_COMMANDS:
- triggers.append(
- {
- **base_trigger,
- CONF_PAYLOAD_TYPE: PAYLOAD_LIGHT_CONTROL_EVENT,
- CONF_TYPE: triggertype,
- CONF_SUBTYPE: light_command,
- }
- )
+ triggers = [
+ {
+ **base_trigger,
+ CONF_PAYLOAD_TYPE: PAYLOAD_LIGHT_CONTROL_EVENT,
+ CONF_TYPE: triggertype,
+ CONF_SUBTYPE: light_command,
+ }
+ for triggertype in TRIGGER_TYPES
+ for light_command in LIGHT_COMMANDS
+ ]
return triggers
diff --git a/custom_components/mlgw/gateway.py b/custom_components/mlgw/gateway.py
index 9d00f9e..719f749 100644
--- a/custom_components/mlgw/gateway.py
+++ b/custom_components/mlgw/gateway.py
@@ -1,21 +1,15 @@
-"""
-The gateway to interact with a Bang & Olufsen MasterLink Gateway or BeoLink Gateway.
-"""
+"""The gateway to interact with a Bang & Olufsen MasterLink Gateway or BeoLink Gateway."""
+
import asyncio
from datetime import datetime
import logging
-
-import telnetlib
import socket
+import telnetlib # need to port to telnetlib3 or exscript
import threading
import time
-from homeassistant.core import callback, HomeAssistant
-from homeassistant.const import (
- STATE_OFF,
- STATE_PLAYING,
- EVENT_HOMEASSISTANT_STOP,
-)
+from homeassistant.const import EVENT_HOMEASSISTANT_STOP, STATE_OFF, STATE_PLAYING
+from homeassistant.core import HomeAssistant, callback
from .const import *
@@ -35,9 +29,9 @@ def __init__(
use_mllog,
default_source,
available_sources,
- hass,
+ hass: HomeAssistant,
config_entry_id=None,
- ):
+ ) -> None:
"""Initialize the MLGW gateway."""
# for both connections
self._host = host
@@ -107,12 +101,14 @@ def set_devices(self, devices):
self._devices = devices
async def terminate_async(self):
- """Terminate the gateway connections. Sets a flag that the listen threads look
- at to determine when to quit"""
+ """Terminate the gateway connections.
+
+ Sets a flag that the listen threads look at to determine when to quit.
+ """
self.stopped.set()
async def async_ml_connect(self):
- """Async version of the mlgw connect function"""
+ """Async version of the mlgw connect function."""
loop = asyncio.get_event_loop()
# start mlgw_connect(self) in a separate thread, suspend
# the current coroutine, and resume when it's done
@@ -181,11 +177,11 @@ def ml_close(self):
# This is the thread function to manage the ML CLI connection
def ml_thread(self):
- """The thread that manages the connection with the MLGW API"""
+ """Manage the connection with the MLGW API."""
connect_retries = 0
max_connect_retries = 10
retry_delay = 60
- while connect_retries < max_connect_retries and not self.stopped.isSet():
+ while connect_retries < max_connect_retries and not self.stopped.is_set():
try:
# if not connected, then connect
if not self._connectedML:
@@ -215,7 +211,7 @@ def ml_listen(self):
_lastping = 0 # how many seconds ago was the last ping.
input_bytes = ""
- while not self.stopped.isSet():
+ while not self.stopped.is_set():
try: # nonblocking read from the connection
input_bytes = input_bytes + self._tn.read_until(
b"\n", _recvtimeout
@@ -227,7 +223,6 @@ def ml_listen(self):
raise
if input_bytes.find("\n") > 0: # if there is a full line
-
line = input_bytes[0 : input_bytes.find("\n")]
input_bytes = input_bytes[input_bytes.find("\n") + 1 :]
@@ -407,7 +402,9 @@ def mlgw_send_beoremoteone_select_source(self, mln, cmd, unit, network_bit: bool
def mlgw_send_beo4_select_source(self, mln, dest, source, sec_source, link):
"""Send Beo4 commmand and store the source name.
- Should change to use a source ID."""
+
+ Should change to use a source ID.
+ """
self._beolink_source = _dictsanitize(beo4_commanddict, source).upper()
self.mlgw_send_beo4_cmd(mln, dest, source, sec_source, link)
@@ -426,14 +423,13 @@ def mlgw_get_serial(self):
self.mlgw_send(MLGW_PL.get("REQUEST SERIAL NUMBER"), "")
(_, self._serial) = self.mlgw_receive()
_LOGGER.info("MLGW: Serial number is %s", self._serial) # info
- return
def mlgw_thread(self):
- """The thread that manages the connection with the MLGW API"""
+ """Manage the connection with the MLGW API."""
connect_retries = 0
max_connect_retries = 10
retry_delay = 60
- while connect_retries < max_connect_retries and not self.stopped.isSet():
+ while connect_retries < max_connect_retries and not self.stopped.is_set():
try:
# if not connected, then connect
if not self._connectedMLGW:
@@ -462,12 +458,12 @@ def mlgw_thread(self):
_LOGGER.warning("Shutting down MLGW API thread")
def _mlgw_listen(self):
- """Listen and manage the MLGW connection"""
+ """Listen and manage the MLGW connection."""
_recvtimeout = 5 # timeout recv every 5 sec
_lastping = 0 # how many seconds ago was the last ping.
self._socket.settimeout(_recvtimeout)
- while not (self.stopped.isSet() or self.brokensocket.isSet()):
+ while not (self.stopped.is_set() or self.brokensocket.is_set()):
response = None
try:
response = self._socket.recv(self.buffersize)
@@ -500,7 +496,7 @@ def _mlgw_listen(self):
sourcePosition = _hexword(response[8], response[9])
sourceActivity = _getdictstr(mlgw_sourceactivitydict, response[10])
pictureFormat = _getdictstr(ml_pictureformatdict, response[11])
- decoded = dict()
+ decoded = {}
decoded["payload_type"] = "source_status"
decoded["source_mln"] = sourceMLN
decoded["source"] = beolink_source
@@ -631,7 +627,9 @@ def _mlgw_listen(self):
def mlgw_receive(self):
"""Receive message from MLGW.
- Returns a tuple: (payload type, payload)."""
+
+ Returns a tuple: (payload type, payload).
+ """
if self._connectedMLGW:
try:
_mlgwdata = self._socket.recv(self.buffersize)
@@ -652,6 +650,7 @@ def mlgw_receive(self):
str(_payloadstr),
)
return (_mlgwdata[1], str(_payloadstr))
+ return None
# ########################################################################################
@@ -670,6 +669,7 @@ async def create_mlgw_gateway(
available_sources=None,
):
"""Create the mlgw gateway.
+
Hass: Home Assistant instance
Host / User / Password: is the login information
mlgw_configurationdata: the configuration information taken from the mlgw_pservices.json API
@@ -773,7 +773,9 @@ def decode_device(d):
def decode_ml_to_dict(telegram) -> dict:
"""Convert a binary ML packet into a dict representation of the message.
- telegram: the binary package"""
+
+ telegram: the binary package
+ """
decoded = dict()
decoded["from_device"] = decode_device(telegram[1])
decoded["to_device"] = decode_device(telegram[0])
@@ -917,6 +919,7 @@ def decode_ml_to_dict(telegram) -> dict:
# ########################################################################################
# ##### Decode MLGW Protocol packet to readable string
+
## Get decoded string for mlgw packet's payload type
#
def _getpayloadtypestr(payloadtype):
diff --git a/custom_components/mlgw/manifest.json b/custom_components/mlgw/manifest.json
index bfa84cf..091406b 100644
--- a/custom_components/mlgw/manifest.json
+++ b/custom_components/mlgw/manifest.json
@@ -21,11 +21,11 @@
"issue_tracker": "https://github.com/giachello/mlgw/issues",
"requirements": [],
"ssdp": [],
- "version" : "2024.8.1",
+ "version" : "2024.8.2",
"zeroconf": [
{
- "name": "*lgw*"
- "type": "_http._tcp.local.",
+ "name": "*lgw*",
+ "type": "_http._tcp.local."
}
]
}
diff --git a/custom_components/mlgw/media_player.py b/custom_components/mlgw/media_player.py
index cdbc461..f2b066a 100644
--- a/custom_components/mlgw/media_player.py
+++ b/custom_components/mlgw/media_player.py
@@ -1,6 +1,4 @@
-"""
-
-Media Player platform for Master Link Gateway connected devices.
+"""Media Player platform for Master Link Gateway connected devices.
------------------------------------------------------------
Where the current sources get modified. There are 3 places:
@@ -30,39 +28,32 @@
"""
-from homeassistant.config_entries import ConfigEntry
-from homeassistant.core import HomeAssistant
-from homeassistant.core import Event, CALLBACK_TYPE
-import logging
-from homeassistant.helpers.entity import DeviceInfo
-import asyncio
-from homeassistant.const import (
- STATE_OFF,
- STATE_PLAYING,
- STATE_PAUSED,
-)
+import asyncio
+import logging
from homeassistant.components.media_player import (
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaType,
)
-
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import STATE_OFF, STATE_PAUSED, STATE_PLAYING
+from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant
+from homeassistant.helpers.entity import DeviceInfo
from .const import (
- DOMAIN,
- reverse_ml_destselectordict,
- reverse_ml_selectedsourcedict,
- ml_selectedsourcedict,
- ml_selectedsource_type_dict,
BEO4_CMDS,
+ DOMAIN,
+ ML_ID_TIMEOUT,
+ MLGW_EVENT_ML_TELEGRAM,
MLGW_GATEWAY,
MLGW_GATEWAY_CONFIGURATION_DATA,
- MLGW_EVENT_ML_TELEGRAM,
- ML_ID_TIMEOUT,
+ ml_selectedsource_type_dict,
+ ml_selectedsourcedict,
+ reverse_ml_destselectordict,
+ reverse_ml_selectedsourcedict,
)
-
from .gateway import MasterLinkGateway
SUPPORT_BEO = (
@@ -80,6 +71,7 @@
# Set up the Media_player devices. there are two ways, through the manual configuration in configuration.yaml
# and through a config flow that automatically reads the devices list from the mlgw.
+
# #########################################################################################
#
async def async_setup_entry(
@@ -87,7 +79,7 @@ async def async_setup_entry(
config_entry: ConfigEntry,
async_add_entities,
):
- """Add MLGW devices through Config Entry"""
+ """Add MLGW devices through Config Entry."""
hass.data.setdefault(DOMAIN, {})
@@ -106,7 +98,7 @@ async def async_setup_entry(
# #########################################################################################
#
async def async_setup_platform(hass, config, add_devices, discovery_info=None):
- """Add MLGW devices through manual configuration"""
+ """Add MLGW devices through manual configuration."""
hass.data.setdefault(DOMAIN, {})
mlgw_configurationdata = hass.data[DOMAIN][MLGW_GATEWAY_CONFIGURATION_DATA]
@@ -120,10 +112,10 @@ async def async_setup_platform(hass, config, add_devices, discovery_info=None):
async def async_create_devices(
mlgw_configurationdata, gateway, async_add_entities, serial=""
):
- """Read the configuration data from the gateway, and create the devices"""
- mp_devices = list()
+ """Read the configuration data from the gateway, and create the devices."""
+ mp_devices = []
- device_sequence = list()
+ device_sequence = []
ml_listener_iteration: int = 0
ml_devices_scanned: int = 0
@@ -147,7 +139,6 @@ def _message_listener(_event: Event):
ml_listener_iteration = ml_listener_iteration + 1
if gateway.connectedMLGW:
-
# listen to ML messages to track down the actual ML id of the device
if gateway.use_mllog:
stop_listening = gateway._hass.bus.async_listen(
@@ -156,9 +147,7 @@ def _message_listener(_event: Event):
for zone in mlgw_configurationdata["zones"]:
for product in zone["products"]:
- device_source_names = list()
- for source in product["sources"]:
- device_source_names.append(source["name"])
+ device_source_names = [source["name"] for source in product["sources"]]
beospeaker = BeoSpeaker(
product["MLN"],
product["name"],
@@ -209,7 +198,7 @@ def _message_listener(_event: Event):
def statusID_to_selectID(statusId):
- """Convert statusID into selectID (e.g., Radio 0x6f ==> 0x81)"""
+ """Convert statusID into selectID (e.g., Radio 0x6f ==> 0x81)."""
return BEO4_CMDS.get(ml_selectedsourcedict.get(statusId).upper())
@@ -238,7 +227,7 @@ def __init__(
source_names: list,
sources: list,
serial="",
- ):
+ ) -> None:
self._mln = mln
self._ml = None
self._name = name
@@ -432,12 +421,12 @@ def clear_media_info(self):
self._media_image_url = None
def set_source_info(self, sourceID, channel_track=0):
- """fill in channel number, name and icon for the UI, if the source ID matches the current source"""
+ """Fill in channel number, name and icon for the UI, if the source ID matches the current source."""
if self._source is None:
return
_statusID = self._sources[self._source_names.index(self._source)]["statusID"]
if _statusID == sourceID:
- if self._playing == False:
+ if not self._playing:
self._playing = True
if channel_track == self._media_channel: # no cange
@@ -471,7 +460,7 @@ def set_source_info(self, sourceID, channel_track=0):
self.schedule_update_ha_state()
def ch_number_to_name_and_icon(self, source, channel_track):
- """look up the caption corresponding to the command number of the favorites list"""
+ """look up the caption corresponding to the command number of the favorites list."""
try:
source_info = self._sources[self._source_names.index(source)]
if "channels" in source_info: # check if the source has favorites
@@ -609,7 +598,7 @@ def set_ml(self, ml: str):
self._ml = ml
def set_state(self, _state):
- # to be called by the gateway to set the state to off when there is an event on the ml bus that turns off the device
+ """To be called by the gateway to set the state to off when there is an event on the ml bus that turns off the device."""
if _state == STATE_PLAYING:
self._pwon = True
self._playing = True
@@ -619,7 +608,7 @@ def set_state(self, _state):
self.clear_media_info()
def set_source(self, source):
- """to be called by the gateway to set the source (the source is a statusID e.g., radio=0x6f)"""
+ """To be called by the gateway to set the source (the source is a statusID e.g., radio=0x6f)."""
# find the source based on the source ID
for _x in self._sources:
if _x["statusID"] == source or _x["selectID"] == statusID_to_selectID(
@@ -634,10 +623,13 @@ def set_source(self, source):
)
def turn_on(self):
- # when turning on this speaker, use the last known source active on beolink
- # if there is no such source, then use the last source used on this speaker
- # if there is no such source, then use the first source in the available sources list.
- # if there is no source in that list, then do nothing
+ """Turn on speaker.
+
+ When turning on this speaker, use the last known source active on beolink
+ If there is no such source, then use the last source used on this speaker
+ If there is no such source, then use the first source in the available sources list.
+ If there is no source in that list, then do nothing
+ """
if self._gateway.beolink_source is not None:
for _x in self._sources:
if _x["statusID"] == reverse_ml_selectedsourcedict.get(
@@ -653,11 +645,11 @@ def turn_on(self):
if len(self._source_names) > 0:
self.select_source(self._source_names[0])
_LOGGER.debug(
- "BeoSpeaker: turn on failed %s %s %s"
- % (self._gateway.beolink_source, self._source, self._source_names[0])
+ f"BeoSpeaker: turn on failed {self._gateway.beolink_source} {self._source} {self._source_names[0]}" # noqa: G004
)
def turn_off(self):
+ """Turn off speaker."""
self._pwon = False
self._playing = False
self.clear_media_info()
@@ -668,7 +660,7 @@ def turn_off(self):
)
def select_source(self, source):
- # look up the full information record for the source
+ """Look up the full information record for the source."""
try:
_LOGGER.debug("BeoSpeaker: trying to select source: %s", source)
source_info = self._sources[self._source_names.index(source)]
@@ -705,6 +697,7 @@ def select_source(self, source):
_LOGGER.debug("BeoSpeaker: source not known: %s", source)
def volume_up(self):
+ """Crank up the volume."""
dest = self._sources[self._source_names.index(self._source)]["destination"]
self._gateway.mlgw_send_beo4_cmd(
self._mln,
@@ -713,6 +706,7 @@ def volume_up(self):
)
def volume_down(self):
+ """Turn down the volume."""
dest = self._sources[self._source_names.index(self._source)]["destination"]
self._gateway.mlgw_send_beo4_cmd(
self._mln,
@@ -721,6 +715,7 @@ def volume_down(self):
)
def mute_volume(self, mute):
+ """Mute speaker."""
dest = self._sources[self._source_names.index(self._source)]["destination"]
self._gateway.mlgw_send_beo4_cmd(
self._mln,