mirror of
https://github.com/home-assistant/core.git
synced 2025-07-10 06:47:09 +00:00
Use xmod model info for Shelly XMOD devices (#137013)
This commit is contained in:
parent
ad7780291e
commit
901011de7b
@ -7,7 +7,12 @@ from typing import Any, Final
|
|||||||
|
|
||||||
from aioshelly.block_device import BlockDevice
|
from aioshelly.block_device import BlockDevice
|
||||||
from aioshelly.common import ConnectionOptions, get_info
|
from aioshelly.common import ConnectionOptions, get_info
|
||||||
from aioshelly.const import BLOCK_GENERATIONS, DEFAULT_HTTP_PORT, RPC_GENERATIONS
|
from aioshelly.const import (
|
||||||
|
BLOCK_GENERATIONS,
|
||||||
|
DEFAULT_HTTP_PORT,
|
||||||
|
MODEL_WALL_DISPLAY,
|
||||||
|
RPC_GENERATIONS,
|
||||||
|
)
|
||||||
from aioshelly.exceptions import (
|
from aioshelly.exceptions import (
|
||||||
CustomPortNotSupported,
|
CustomPortNotSupported,
|
||||||
DeviceConnectionError,
|
DeviceConnectionError,
|
||||||
@ -41,7 +46,6 @@ from .const import (
|
|||||||
CONF_SLEEP_PERIOD,
|
CONF_SLEEP_PERIOD,
|
||||||
DOMAIN,
|
DOMAIN,
|
||||||
LOGGER,
|
LOGGER,
|
||||||
MODEL_WALL_DISPLAY,
|
|
||||||
BLEScannerMode,
|
BLEScannerMode,
|
||||||
)
|
)
|
||||||
from .coordinator import async_reconnect_soon
|
from .coordinator import async_reconnect_soon
|
||||||
@ -112,7 +116,7 @@ async def validate_input(
|
|||||||
return {
|
return {
|
||||||
"title": rpc_device.name,
|
"title": rpc_device.name,
|
||||||
CONF_SLEEP_PERIOD: sleep_period,
|
CONF_SLEEP_PERIOD: sleep_period,
|
||||||
"model": rpc_device.shelly.get("model"),
|
"model": rpc_device.xmod_info.get("p") or rpc_device.shelly.get("model"),
|
||||||
CONF_GEN: gen,
|
CONF_GEN: gen,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -10,7 +10,7 @@ from typing import Any, cast
|
|||||||
|
|
||||||
from aioshelly.ble import async_ensure_ble_enabled, async_stop_scanner
|
from aioshelly.ble import async_ensure_ble_enabled, async_stop_scanner
|
||||||
from aioshelly.block_device import BlockDevice, BlockUpdateType
|
from aioshelly.block_device import BlockDevice, BlockUpdateType
|
||||||
from aioshelly.const import MODEL_NAMES, MODEL_VALVE
|
from aioshelly.const import MODEL_VALVE
|
||||||
from aioshelly.exceptions import (
|
from aioshelly.exceptions import (
|
||||||
DeviceConnectionError,
|
DeviceConnectionError,
|
||||||
InvalidAuthError,
|
InvalidAuthError,
|
||||||
@ -68,6 +68,7 @@ from .utils import (
|
|||||||
async_create_issue_unsupported_firmware,
|
async_create_issue_unsupported_firmware,
|
||||||
get_block_device_sleep_period,
|
get_block_device_sleep_period,
|
||||||
get_device_entry_gen,
|
get_device_entry_gen,
|
||||||
|
get_device_info_model,
|
||||||
get_host,
|
get_host,
|
||||||
get_http_port,
|
get_http_port,
|
||||||
get_rpc_device_wakeup_period,
|
get_rpc_device_wakeup_period,
|
||||||
@ -164,7 +165,7 @@ class ShellyCoordinatorBase[_DeviceT: BlockDevice | RpcDevice](
|
|||||||
connections={(CONNECTION_NETWORK_MAC, self.mac)},
|
connections={(CONNECTION_NETWORK_MAC, self.mac)},
|
||||||
identifiers={(DOMAIN, self.mac)},
|
identifiers={(DOMAIN, self.mac)},
|
||||||
manufacturer="Shelly",
|
manufacturer="Shelly",
|
||||||
model=MODEL_NAMES.get(self.model),
|
model=get_device_info_model(self.device),
|
||||||
model_id=self.model,
|
model_id=self.model,
|
||||||
sw_version=self.sw_version,
|
sw_version=self.sw_version,
|
||||||
hw_version=f"gen{get_device_entry_gen(self.config_entry)}",
|
hw_version=f"gen{get_device_entry_gen(self.config_entry)}",
|
||||||
|
@ -315,6 +315,14 @@ def get_model_name(info: dict[str, Any]) -> str:
|
|||||||
return cast(str, MODEL_NAMES.get(info["type"], info["type"]))
|
return cast(str, MODEL_NAMES.get(info["type"], info["type"]))
|
||||||
|
|
||||||
|
|
||||||
|
def get_device_info_model(device: BlockDevice | RpcDevice) -> str | None:
|
||||||
|
"""Return the device model for deviceinfo."""
|
||||||
|
if isinstance(device, RpcDevice) and (model := device.xmod_info.get("n")):
|
||||||
|
return cast(str, model)
|
||||||
|
|
||||||
|
return cast(str, MODEL_NAMES.get(device.model))
|
||||||
|
|
||||||
|
|
||||||
def get_rpc_channel_name(device: RpcDevice, key: str) -> str:
|
def get_rpc_channel_name(device: RpcDevice, key: str) -> str:
|
||||||
"""Get name based on device and channel name."""
|
"""Get name based on device and channel name."""
|
||||||
key = key.replace("emdata", "em")
|
key = key.replace("emdata", "em")
|
||||||
|
@ -476,6 +476,7 @@ def _mock_rpc_device(version: str | None = None):
|
|||||||
script_getcode=AsyncMock(
|
script_getcode=AsyncMock(
|
||||||
side_effect=lambda script_id: {"data": MOCK_SCRIPTS[script_id - 1]}
|
side_effect=lambda script_id: {"data": MOCK_SCRIPTS[script_id - 1]}
|
||||||
),
|
),
|
||||||
|
xmod_info={},
|
||||||
)
|
)
|
||||||
type(device).name = PropertyMock(return_value="Test name")
|
type(device).name = PropertyMock(return_value="Test name")
|
||||||
return device
|
return device
|
||||||
|
@ -1011,3 +1011,22 @@ async def test_rpc_already_connected(
|
|||||||
|
|
||||||
assert "already connected" in caplog.text
|
assert "already connected" in caplog.text
|
||||||
mock_rpc_device.initialize.assert_called_once()
|
mock_rpc_device.initialize.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_xmod_model_lookup(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
device_registry: dr.DeviceRegistry,
|
||||||
|
mock_rpc_device: Mock,
|
||||||
|
monkeypatch: pytest.MonkeyPatch,
|
||||||
|
) -> None:
|
||||||
|
"""Test XMOD model look-up."""
|
||||||
|
xmod_model = "Test XMOD model name"
|
||||||
|
monkeypatch.setattr(mock_rpc_device, "xmod_info", {"n": xmod_model})
|
||||||
|
entry = await init_integration(hass, 2)
|
||||||
|
|
||||||
|
device = device_registry.async_get_device(
|
||||||
|
identifiers={(DOMAIN, entry.entry_id)},
|
||||||
|
connections={(dr.CONNECTION_NETWORK_MAC, dr.format_mac(entry.unique_id))},
|
||||||
|
)
|
||||||
|
assert device
|
||||||
|
assert device.model == xmod_model
|
||||||
|
Loading…
x
Reference in New Issue
Block a user