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

System Bridge 3.x.x (#71218)

Browse source 
* Change to new package and tcp

* Rework integration pt1

* Show by default

* Handle auth error

* Use const

* New version avaliable (to be replaced in future by update entity)

* Remove visible

* Version latest

* Filesystem space use

* Dev package

* Fix sensor

* Add services

* Update package

* Add temperature and voltage

* GPU

* Bump package version

* Update config flow

* Add displays

* Fix displays connected

* Round to whole number

* GPU fan speed in RPM

* Handle disconnections

* Update package

* Fix

* Update tests

* Handle more errors

* Check submodule and return missing uuid in test

* Handle auth error on config flow

* Fix test

* Bump package version

* Handle key errors

* Update package to release version

* Client session in config flow

* Log

* Increase timeout and use similar logic in config flow to init

* 30 secs

* Add test for timeout error

* Cleanup logs

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update tests/components/system_bridge/test_config_flow.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* uuid raise specific error

* Type

* Lambda to functions for complex logic

* Unknown error test

* Bump package to 3.0.5

* Bump package to 3.0.6

* Use typings from package and pydantic

* Use dict()

* Use data listener function and map to models

* Use passed module handler

* Use lists from models

* Update to 3.1.0

* Update coordinator to use passed module

* Improve coordinator

* Add debug

* Bump package and avaliable -> available

* Add version check

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Aidan Timson 2022-06-01 22:54:22 +01:00 • committed by GitHub
parent 05296fb86e
commit 2ba45a9f99
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 892 additions and 731 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

230
homeassistant/components/system_bridge/__init__.py
View file

@ -3,102 +3,114 @@ from __future__ import annotations
import asyncio
import logging
import shlex
import async_timeout
from systembridge import Bridge
from systembridge.client import BridgeClient
from systembridge.exceptions import BridgeAuthenticationException
from systembridge.objects.command.response import CommandResponse
from systembridge.objects.keyboard.payload import KeyboardPayload
from systembridgeconnector.exceptions import (
AuthenticationException,
ConnectionClosedException,
ConnectionErrorException,
)
from systembridgeconnector.version import SUPPORTED_VERSION, Version
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_API_KEY,
CONF_COMMAND,
CONF_HOST,
CONF_PATH,
CONF_PORT,
CONF_URL,
Platform,
)
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryNotReady,
HomeAssistantError,
)
from homeassistant.helpers import (
aiohttp_client,
config_validation as cv,
device_registry as dr,
)
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import BRIDGE_CONNECTION_ERRORS, DOMAIN
from .const import DOMAIN, MODULES
from .coordinator import SystemBridgeDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.SENSOR,
]
CONF_ARGUMENTS = "arguments"
CONF_BRIDGE = "bridge"
CONF_KEY = "key"
CONF_MODIFIERS = "modifiers"
CONF_TEXT = "text"
CONF_WAIT = "wait"
SERVICE_SEND_COMMAND = "send_command"
SERVICE_OPEN = "open"
SERVICE_OPEN_PATH = "open_path"
SERVICE_OPEN_URL = "open_url"
SERVICE_SEND_KEYPRESS = "send_keypress"
SERVICE_SEND_TEXT = "send_text"
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up System Bridge from a config entry."""
bridge = Bridge(
BridgeClient(aiohttp_client.async_get_clientsession(hass)),
f"http://{entry.data[CONF_HOST]}:{entry.data[CONF_PORT]}",
entry.data[CONF_API_KEY],
)
# Check version before initialising
version = Version(
entry.data[CONF_HOST],
entry.data[CONF_PORT],
entry.data[CONF_API_KEY],
session=async_get_clientsession(hass),
)
try:
async with async_timeout.timeout(30):
await bridge.async_get_information()
except BridgeAuthenticationException as exception:
raise ConfigEntryAuthFailed(
f"Authentication failed for {entry.title} ({entry.data[CONF_HOST]})"
) from exception
except BRIDGE_CONNECTION_ERRORS as exception:
if not await version.check_supported():
raise ConfigEntryNotReady(
f"You are not running a supported version of System Bridge. Please update to {SUPPORTED_VERSION} or higher."
)
except AuthenticationException as exception:
_LOGGER.error("Authentication failed for %s: %s", entry.title, exception)
raise ConfigEntryAuthFailed from exception
except (ConnectionClosedException, ConnectionErrorException) as exception:
raise ConfigEntryNotReady(
f"Could not connect to {entry.title} ({entry.data[CONF_HOST]})."
) from exception
except asyncio.TimeoutError as exception:
raise ConfigEntryNotReady(
f"Timed out waiting for {entry.title} ({entry.data[CONF_HOST]})."
) from exception
coordinator = SystemBridgeDataUpdateCoordinator(
hass,
_LOGGER,
entry=entry,
)
try:
async with async_timeout.timeout(30):
await coordinator.async_get_data(MODULES)
except AuthenticationException as exception:
_LOGGER.error("Authentication failed for %s: %s", entry.title, exception)
raise ConfigEntryAuthFailed from exception
except (ConnectionClosedException, ConnectionErrorException) as exception:
raise ConfigEntryNotReady(
f"Could not connect to {entry.title} ({entry.data[CONF_HOST]})."
) from exception
except asyncio.TimeoutError as exception:
raise ConfigEntryNotReady(
f"Timed out waiting for {entry.title} ({entry.data[CONF_HOST]})."
) from exception
coordinator = SystemBridgeDataUpdateCoordinator(hass, bridge, _LOGGER, entry=entry)
await coordinator.async_config_entry_first_refresh()
# Wait for initial data
_LOGGER.debug("Data: %s", coordinator.data)
try:
async with async_timeout.timeout(60):
while (
coordinator.bridge.battery is None
or coordinator.bridge.cpu is None
or coordinator.bridge.display is None
or coordinator.bridge.filesystem is None
or coordinator.bridge.graphics is None
or coordinator.bridge.information is None
or coordinator.bridge.memory is None
or coordinator.bridge.network is None
or coordinator.bridge.os is None
or coordinator.bridge.processes is None
or coordinator.bridge.system is None
# Wait for initial data
async with async_timeout.timeout(30):
while coordinator.data is None or all(
getattr(coordinator.data, module) is None for module in MODULES
):
_LOGGER.debug(
"Waiting for initial data from %s (%s)",
"Waiting for initial data from %s (%s): %s",
entry.title,
entry.data[CONF_HOST],
coordinator.data,
)
await asyncio.sleep(1)
except asyncio.TimeoutError as exception:
@ -111,7 +123,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
if hass.services.has_service(DOMAIN, SERVICE_SEND_COMMAND):
if hass.services.has_service(DOMAIN, SERVICE_OPEN_URL):
return True
def valid_device(device: str):
@ -129,104 +141,56 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
raise vol.Invalid from exception
raise vol.Invalid(f"Device {device} does not exist")
async def handle_send_command(call: ServiceCall) -> None:
"""Handle the send_command service call."""
async def handle_open_path(call: ServiceCall) -> None:
"""Handle the open path service call."""
_LOGGER.info("Open: %s", call.data)
coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][
call.data[CONF_BRIDGE]
]
bridge: Bridge = coordinator.bridge
await coordinator.websocket_client.open_path(call.data[CONF_PATH])
command = call.data[CONF_COMMAND]
arguments = shlex.split(call.data[CONF_ARGUMENTS])
_LOGGER.debug(
"Command payload: %s",
{CONF_COMMAND: command, CONF_ARGUMENTS: arguments, CONF_WAIT: False},
)
try:
response: CommandResponse = await bridge.async_send_command(
{CONF_COMMAND: command, CONF_ARGUMENTS: arguments, CONF_WAIT: False}
)
if not response.success:
raise HomeAssistantError(
f"Error sending command. Response message was: {response.message}"
)
except (BridgeAuthenticationException, *BRIDGE_CONNECTION_ERRORS) as exception:
raise HomeAssistantError("Error sending command") from exception
_LOGGER.debug("Sent command. Response message was: %s", response.message)
async def handle_open(call: ServiceCall) -> None:
"""Handle the open service call."""
async def handle_open_url(call: ServiceCall) -> None:
"""Handle the open url service call."""
_LOGGER.info("Open: %s", call.data)
coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][
call.data[CONF_BRIDGE]
]
bridge: Bridge = coordinator.bridge
path = call.data[CONF_PATH]
_LOGGER.debug("Open payload: %s", {CONF_PATH: path})
try:
await bridge.async_open({CONF_PATH: path})
except (BridgeAuthenticationException, *BRIDGE_CONNECTION_ERRORS) as exception:
raise HomeAssistantError("Error sending") from exception
_LOGGER.debug("Sent open request")
await coordinator.websocket_client.open_url(call.data[CONF_URL])
async def handle_send_keypress(call: ServiceCall) -> None:
"""Handle the send_keypress service call."""
coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][
call.data[CONF_BRIDGE]
]
bridge: Bridge = coordinator.data
keyboard_payload: KeyboardPayload = {
CONF_KEY: call.data[CONF_KEY],
CONF_MODIFIERS: shlex.split(call.data.get(CONF_MODIFIERS, "")),
}
_LOGGER.debug("Keypress payload: %s", keyboard_payload)
try:
await bridge.async_send_keypress(keyboard_payload)
except (BridgeAuthenticationException, *BRIDGE_CONNECTION_ERRORS) as exception:
raise HomeAssistantError("Error sending") from exception
_LOGGER.debug("Sent keypress request")
await coordinator.websocket_client.keyboard_keypress(call.data[CONF_KEY])
async def handle_send_text(call: ServiceCall) -> None:
"""Handle the send_keypress service call."""
coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][
call.data[CONF_BRIDGE]
]
bridge: Bridge = coordinator.data
keyboard_payload: KeyboardPayload = {CONF_TEXT: call.data[CONF_TEXT]}
_LOGGER.debug("Text payload: %s", keyboard_payload)
try:
await bridge.async_send_keypress(keyboard_payload)
except (BridgeAuthenticationException, *BRIDGE_CONNECTION_ERRORS) as exception:
raise HomeAssistantError("Error sending") from exception
_LOGGER.debug("Sent text request")
await coordinator.websocket_client.keyboard_text(call.data[CONF_TEXT])
hass.services.async_register(
DOMAIN,
SERVICE_SEND_COMMAND,
handle_send_command,
SERVICE_OPEN_PATH,
handle_open_path,
schema=vol.Schema(
{
vol.Required(CONF_BRIDGE): valid_device,
vol.Required(CONF_COMMAND): cv.string,
vol.Optional(CONF_ARGUMENTS, ""): cv.string,
vol.Required(CONF_PATH): cv.string,
},
),
)
hass.services.async_register(
DOMAIN,
SERVICE_OPEN,
handle_open,
SERVICE_OPEN_URL,
handle_open_url,
schema=vol.Schema(
{
vol.Required(CONF_BRIDGE): valid_device,
vol.Required(CONF_PATH): cv.string,
vol.Required(CONF_URL): cv.string,
},
),
)
@ -239,7 +203,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
{
vol.Required(CONF_BRIDGE): valid_device,
vol.Required(CONF_KEY): cv.string,
vol.Optional(CONF_MODIFIERS): cv.string,
},
),
)
@ -271,15 +234,17 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
]
# Ensure disconnected and cleanup stop sub
await coordinator.bridge.async_close_websocket()
await coordinator.websocket_client.close()
if coordinator.unsub:
coordinator.unsub()
del hass.data[DOMAIN][entry.entry_id]
if not hass.data[DOMAIN]:
hass.services.async_remove(DOMAIN, SERVICE_SEND_COMMAND)
hass.services.async_remove(DOMAIN, SERVICE_OPEN)
hass.services.async_remove(DOMAIN, SERVICE_OPEN_PATH)
hass.services.async_remove(DOMAIN, SERVICE_OPEN_URL)
hass.services.async_remove(DOMAIN, SERVICE_SEND_KEYPRESS)
hass.services.async_remove(DOMAIN, SERVICE_SEND_TEXT)
return unload_ok
@ -295,20 +260,21 @@ class SystemBridgeEntity(CoordinatorEntity[SystemBridgeDataUpdateCoordinator]):
def __init__(
self,
coordinator: SystemBridgeDataUpdateCoordinator,
api_port: int,
key: str,
name: str | None,
) -> None:
"""Initialize the System Bridge entity."""
super().__init__(coordinator)
bridge: Bridge = coordinator.data
self._key = f"{bridge.information.host}_{key}"
self._name = f"{bridge.information.host} {name}"
self._configuration_url = bridge.get_configuration_url()
self._hostname = bridge.information.host
self._mac = bridge.information.mac
self._manufacturer = bridge.system.system.manufacturer
self._model = bridge.system.system.model
self._version = bridge.system.system.version
self._hostname = coordinator.data.system.hostname
self._key = f"{self._hostname}_{key}"
self._name = f"{self._hostname} {name}"
self._configuration_url = (
f"http://{self._hostname}:{api_port}/app/settings.html"
)
self._mac_address = coordinator.data.system.mac_address
self._version = coordinator.data.system.version
@property
def unique_id(self) -> str:
@ -329,9 +295,7 @@ class SystemBridgeDeviceEntity(SystemBridgeEntity):
"""Return device information about this System Bridge instance."""
return DeviceInfo(
configuration_url=self._configuration_url,
connections={(dr.CONNECTION_NETWORK_MAC, self._mac)},
manufacturer=self._manufacturer,
model=self._model,
connections={(dr.CONNECTION_NETWORK_MAC, self._mac_address)},
name=self._hostname,
sw_version=self._version,
)

29
homeassistant/components/system_bridge/binary_sensor.py
View file

@ -4,14 +4,13 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from systembridge import Bridge
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PORT
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -32,7 +31,7 @@ BASE_BINARY_SENSOR_TYPES: tuple[SystemBridgeBinarySensorEntityDescription, ...]
key="version_available",
name="New Version Available",
device_class=BinarySensorDeviceClass.UPDATE,
value=lambda bridge: bridge.information.updates.available,
value=lambda data: data.system.version_newer_available,
),
)
@ -41,7 +40,7 @@ BATTERY_BINARY_SENSOR_TYPES: tuple[SystemBridgeBinarySensorEntityDescription, ..
key="battery_is_charging",
name="Battery Is Charging",
device_class=BinarySensorDeviceClass.BATTERY_CHARGING,
value=lambda bridge: bridge.battery.isCharging,
value=lambda data: data.battery.is_charging,
),
)
@ -51,15 +50,24 @@ async def async_setup_entry(
) -> None:
"""Set up System Bridge binary sensor based on a config entry."""
coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
bridge: Bridge = coordinator.data
entities = []
for description in BASE_BINARY_SENSOR_TYPES:
entities.append(SystemBridgeBinarySensor(coordinator, description))
entities.append(
SystemBridgeBinarySensor(coordinator, description, entry.data[CONF_PORT])
)
if bridge.battery and bridge.battery.hasBattery:
if (
coordinator.data.battery
and coordinator.data.battery.percentage
and coordinator.data.battery.percentage > -1
):
for description in BATTERY_BINARY_SENSOR_TYPES:
entities.append(SystemBridgeBinarySensor(coordinator, description))
entities.append(
SystemBridgeBinarySensor(
coordinator, description, entry.data[CONF_PORT]
)
)
async_add_entities(entities)
@ -73,10 +81,12 @@ class SystemBridgeBinarySensor(SystemBridgeDeviceEntity, BinarySensorEntity):
self,
coordinator: SystemBridgeDataUpdateCoordinator,
description: SystemBridgeBinarySensorEntityDescription,
api_port: int,
) -> None:
"""Initialize."""
super().__init__(
coordinator,
api_port,
description.key,
description.name,
)
@ -85,5 +95,4 @@ class SystemBridgeBinarySensor(SystemBridgeDeviceEntity, BinarySensorEntity):
@property
def is_on(self) -> bool:
"""Return the boolean state of the binary sensor."""
bridge: Bridge = self.coordinator.data
return self.entity_description.value(bridge)
return self.entity_description.value(self.coordinator.data)

124
homeassistant/components/system_bridge/config_flow.py
View file

@ -1,13 +1,18 @@
"""Config flow for System Bridge integration."""
from __future__ import annotations
import asyncio
import logging
from typing import Any
import async_timeout
from systembridge import Bridge
from systembridge.client import BridgeClient
from systembridge.exceptions import BridgeAuthenticationException
from systembridgeconnector.const import EVENT_MODULE, EVENT_TYPE, TYPE_DATA_UPDATE
from systembridgeconnector.exceptions import (
AuthenticationException,
ConnectionClosedException,
ConnectionErrorException,
)
from systembridgeconnector.websocket_client import WebSocketClient
import voluptuous as vol
from homeassistant import config_entries, exceptions
@ -15,9 +20,10 @@ from homeassistant.components import zeroconf
from homeassistant.const import CONF_API_KEY, CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client, config_validation as cv
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import BRIDGE_CONNECTION_ERRORS, DOMAIN
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -31,39 +37,84 @@ STEP_USER_DATA_SCHEMA = vol.Schema(
)
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, str]:
async def validate_input(
hass: HomeAssistant,
data: dict[str, Any],
) -> dict[str, str]:
"""Validate the user input allows us to connect.
Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
"""
bridge = Bridge(
BridgeClient(aiohttp_client.async_get_clientsession(hass)),
f"http://{data[CONF_HOST]}:{data[CONF_PORT]}",
host = data[CONF_HOST]
websocket_client = WebSocketClient(
host,
data[CONF_PORT],
data[CONF_API_KEY],
)
hostname = data[CONF_HOST]
try:
async with async_timeout.timeout(30):
await bridge.async_get_information()
if (
bridge.information is not None
and bridge.information.host is not None
and bridge.information.uuid is not None
):
hostname = bridge.information.host
uuid = bridge.information.uuid
except BridgeAuthenticationException as exception:
_LOGGER.info(exception)
await websocket_client.connect(session=async_get_clientsession(hass))
await websocket_client.get_data(["system"])
while True:
message = await websocket_client.receive_message()
_LOGGER.debug("Message: %s", message)
if (
message[EVENT_TYPE] == TYPE_DATA_UPDATE
and message[EVENT_MODULE] == "system"
):
break
except AuthenticationException as exception:
_LOGGER.warning(
"Authentication error when connecting to %s: %s", data[CONF_HOST], exception
)
raise InvalidAuth from exception
except BRIDGE_CONNECTION_ERRORS as exception:
_LOGGER.info(exception)
except (
ConnectionClosedException,
ConnectionErrorException,
) as exception:
_LOGGER.warning(
"Connection error when connecting to %s: %s", data[CONF_HOST], exception
)
raise CannotConnect from exception
except asyncio.TimeoutError as exception:
_LOGGER.warning("Timed out connecting to %s: %s", data[CONF_HOST], exception)
raise CannotConnect from exception
return {"hostname": hostname, "uuid": uuid}
_LOGGER.debug("%s Message: %s", TYPE_DATA_UPDATE, message)
if "uuid" not in message["data"]:
error = "No UUID in result!"
raise CannotConnect(error)
return {"hostname": host, "uuid": message["data"]["uuid"]}
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_get_info(
hass: HomeAssistant,
user_input: dict[str, Any],
) -> tuple[dict[str, str], dict[str, str] | None]:
errors = {}
try:
info = await validate_input(hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return errors, info
return errors, None
class ConfigFlow(
config_entries.ConfigFlow,
domain=DOMAIN,
):
"""Handle a config flow for System Bridge."""
VERSION = 1
@ -74,25 +125,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._input: dict[str, Any] = {}
self._reauth = False
async def _async_get_info(
self, user_input: dict[str, Any]
) -> tuple[dict[str, str], dict[str, str] | None]:
errors = {}
try:
info = await validate_input(self.hass, user_input)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
return errors, info
return errors, None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
@ -102,7 +134,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
errors, info = await self._async_get_info(user_input)
errors, info = await _async_get_info(self.hass, user_input)
if not errors and info is not None:
# Check if already configured
await self.async_set_unique_id(info["uuid"], raise_on_progress=False)
@ -122,7 +154,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if user_input is not None:
user_input = {**self._input, **user_input}
errors, info = await self._async_get_info(user_input)
errors, info = await _async_get_info(self.hass, user_input)
if not errors and info is not None:
# Check if already configured
existing_entry = await self.async_set_unique_id(info["uuid"])

25
homeassistant/components/system_bridge/const.py
View file

@ -1,20 +1,13 @@
"""Constants for the System Bridge integration."""
import asyncio
from aiohttp.client_exceptions import (
ClientConnectionError,
ClientConnectorError,
ClientResponseError,
)
from systembridge.exceptions import BridgeException
DOMAIN = "system_bridge"
BRIDGE_CONNECTION_ERRORS = (
asyncio.TimeoutError,
BridgeException,
ClientConnectionError,
ClientConnectorError,
ClientResponseError,
OSError,
)
MODULES = [
"battery",
"cpu",
"disk",
"display",
"gpu",
"memory",
"system",
]

205
homeassistant/components/system_bridge/coordinator.py
View file

@ -6,40 +6,71 @@ from collections.abc import Callable
from datetime import timedelta
import logging
from systembridge import Bridge
from systembridge.exceptions import (
BridgeAuthenticationException,
BridgeConnectionClosedException,
BridgeException,
import async_timeout
from pydantic import BaseModel # pylint: disable=no-name-in-module
from systembridgeconnector.exceptions import (
AuthenticationException,
ConnectionClosedException,
ConnectionErrorException,
)
from systembridge.objects.events import Event
from systembridgeconnector.models.battery import Battery
from systembridgeconnector.models.cpu import Cpu
from systembridgeconnector.models.disk import Disk
from systembridgeconnector.models.display import Display
from systembridgeconnector.models.gpu import Gpu
from systembridgeconnector.models.memory import Memory
from systembridgeconnector.models.system import System
from systembridgeconnector.websocket_client import WebSocketClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, EVENT_HOMEASSISTANT_STOP
from homeassistant.const import (
CONF_API_KEY,
CONF_HOST,
CONF_PORT,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import BRIDGE_CONNECTION_ERRORS, DOMAIN
from .const import DOMAIN, MODULES
class SystemBridgeDataUpdateCoordinator(DataUpdateCoordinator[Bridge]):
class SystemBridgeCoordinatorData(BaseModel):
"""System Bridge Coordianator Data."""
battery: Battery = None
cpu: Cpu = None
disk: Disk = None
display: Display = None
gpu: Gpu = None
memory: Memory = None
system: System = None
class SystemBridgeDataUpdateCoordinator(
DataUpdateCoordinator[SystemBridgeCoordinatorData]
):
"""Class to manage fetching System Bridge data from single endpoint."""
def __init__(
self,
hass: HomeAssistant,
bridge: Bridge,
LOGGER: logging.Logger,
*,
entry: ConfigEntry,
) -> None:
"""Initialize global System Bridge data updater."""
self.bridge = bridge
self.title = entry.title
self.host = entry.data[CONF_HOST]
self.unsub: Callable | None = None
self.systembridge_data = SystemBridgeCoordinatorData()
self.websocket_client = WebSocketClient(
entry.data[CONF_HOST],
entry.data[CONF_PORT],
entry.data[CONF_API_KEY],
)
super().__init__(
hass, LOGGER, name=DOMAIN, update_interval=timedelta(seconds=30)
)
@ -49,97 +80,117 @@ class SystemBridgeDataUpdateCoordinator(DataUpdateCoordinator[Bridge]):
for update_callback in self._listeners:
update_callback()
async def async_handle_event(self, event: Event):
"""Handle System Bridge events from the WebSocket."""
# No need to update anything, as everything is updated in the caller
self.logger.debug(
"New event from %s (%s): %s", self.title, self.host, event.name
)
self.async_set_updated_data(self.bridge)
async def async_get_data(
self,
modules: list[str],
) -> None:
"""Get data from WebSocket."""
if not self.websocket_client.connected:
await self._setup_websocket()
async def _listen_for_events(self) -> None:
await self.websocket_client.get_data(modules)
async def async_handle_module(
self,
module_name: str,
module,
) -> None:
"""Handle data from the WebSocket client."""
self.logger.debug("Set new data for: %s", module_name)
setattr(self.systembridge_data, module_name, module)
self.async_set_updated_data(self.systembridge_data)
async def _listen_for_data(self) -> None:
"""Listen for events from the WebSocket."""
try:
await self.bridge.async_send_event(
"get-data",
[
{"service": "battery", "method": "findAll", "observe": True},
{"service": "cpu", "method": "findAll", "observe": True},
{"service": "display", "method": "findAll", "observe": True},
{"service": "filesystem", "method": "findSizes", "observe": True},
{"service": "graphics", "method": "findAll", "observe": True},
{"service": "memory", "method": "findAll", "observe": True},
{"service": "network", "method": "findAll", "observe": True},
{"service": "os", "method": "findAll", "observe": False},
{
"service": "processes",
"method": "findCurrentLoad",
"observe": True,
},
{"service": "system", "method": "findAll", "observe": False},
],
)
await self.bridge.listen_for_events(callback=self.async_handle_event)
except BridgeConnectionClosedException as exception:
await self.websocket_client.register_data_listener(MODULES)
await self.websocket_client.listen(callback=self.async_handle_module)
except AuthenticationException as exception:
self.last_update_success = False
self.logger.info(
"Websocket Connection Closed for %s (%s). Will retry: %s",
self.title,
self.host,
exception,
)
except BridgeException as exception:
self.logger.error("Authentication failed for %s: %s", self.title, exception)
if self.unsub:
self.unsub()
self.unsub = None
self.last_update_success = False
self.update_listeners()
self.logger.warning(
"Exception occurred for %s (%s). Will retry: %s",
except (ConnectionClosedException, ConnectionResetError) as exception:
self.logger.info(
"Websocket connection closed for %s. Will retry: %s",
self.title,
self.host,
exception,
)
if self.unsub:
self.unsub()
self.unsub = None
self.last_update_success = False
self.update_listeners()
except ConnectionErrorException as exception:
self.logger.warning(
"Connection error occurred for %s. Will retry: %s",
self.title,
exception,
)
if self.unsub:
self.unsub()
self.unsub = None
self.last_update_success = False
self.update_listeners()
async def _setup_websocket(self) -> None:
"""Use WebSocket for updates."""
try:
self.logger.debug(
"Connecting to ws://%s:%s",
self.host,
self.bridge.information.websocketPort,
)
await self.bridge.async_connect_websocket(
self.host, self.bridge.information.websocketPort
)
except BridgeAuthenticationException as exception:
async with async_timeout.timeout(20):
await self.websocket_client.connect(
session=async_get_clientsession(self.hass),
)
except AuthenticationException as exception:
self.last_update_success = False
self.logger.error("Authentication failed for %s: %s", self.title, exception)
if self.unsub:
self.unsub()
self.unsub = None
raise ConfigEntryAuthFailed() from exception
except (*BRIDGE_CONNECTION_ERRORS, ConnectionRefusedError) as exception:
if self.unsub:
self.unsub()
self.unsub = None
raise UpdateFailed(
f"Could not connect to {self.title} ({self.host})."
) from exception
asyncio.create_task(self._listen_for_events())
self.last_update_success = False
self.update_listeners()
except ConnectionErrorException as exception:
self.logger.warning(
"Connection error occurred for %s. Will retry: %s",
self.title,
exception,
)
self.last_update_success = False
self.update_listeners()
except asyncio.TimeoutError as exception:
self.logger.warning(
"Timed out waiting for %s. Will retry: %s",
self.title,
exception,
)
self.last_update_success = False
self.update_listeners()
self.hass.async_create_task(self._listen_for_data())
self.last_update_success = True
self.update_listeners()
async def close_websocket(_) -> None:
"""Close WebSocket connection."""
await self.bridge.async_close_websocket()
await self.websocket_client.close()
# Clean disconnect WebSocket on Home Assistant shutdown
self.unsub = self.hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, close_websocket
)
async def _async_update_data(self) -> Bridge:
async def _async_update_data(self) -> SystemBridgeCoordinatorData:
"""Update System Bridge data from WebSocket."""
self.logger.debug(
"_async_update_data - WebSocket Connected: %s",
self.bridge.websocket_connected,
self.websocket_client.connected,
)
if not self.bridge.websocket_connected:
if not self.websocket_client.connected:
await self._setup_websocket()
return self.bridge
self.logger.debug("_async_update_data done")
return self.systembridge_data

6
homeassistant/components/system_bridge/manifest.json
View file

@ -3,11 +3,11 @@
"name": "System Bridge",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/system_bridge",
"requirements": ["systembridge==2.3.1"],
"requirements": ["systembridgeconnector==3.1.3"],
"codeowners": ["@timmo001"],
"zeroconf": ["_system-bridge._udp.local."],
"zeroconf": ["_system-bridge._tcp.local."],
"after_dependencies": ["zeroconf"],
"quality_scale": "silver",
"iot_class": "local_push",
"loggers": ["systembridge"]
"loggers": ["systembridgeconnector"]
}

569
homeassistant/components/system_bridge/sensor.py
View file

@ -6,8 +6,6 @@ from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Final, cast
from systembridge import Bridge
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
@ -16,6 +14,7 @@ from homeassistant.components.sensor import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_PORT,
DATA_GIGABYTES,
ELECTRIC_POTENTIAL_VOLT,
FREQUENCY_GIGAHERTZ,
@ -28,10 +27,11 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.util.dt import utcnow
from . import SystemBridgeDeviceEntity
from .const import DOMAIN
from .coordinator import SystemBridgeDataUpdateCoordinator
from .coordinator import SystemBridgeCoordinatorData, SystemBridgeDataUpdateCoordinator
ATTR_AVAILABLE: Final = "available"
ATTR_FILESYSTEM: Final = "filesystem"
@ -41,6 +41,7 @@ ATTR_TYPE: Final = "type"
ATTR_USED: Final = "used"
PIXELS: Final = "px"
RPM: Final = "RPM"
@dataclass
@ -50,21 +51,87 @@ class SystemBridgeSensorEntityDescription(SensorEntityDescription):
value: Callable = round
def battery_time_remaining(data: SystemBridgeCoordinatorData) -> datetime | None:
"""Return the battery time remaining."""
if data.battery.sensors_secsleft is not None:
return utcnow() + timedelta(seconds=data.battery.sensors_secsleft)
return None
def cpu_speed(data: SystemBridgeCoordinatorData) -> float | None:
"""Return the CPU speed."""
if data.cpu.frequency_current is not None:
return round(data.cpu.frequency_current / 1000, 2)
return None
def gpu_core_clock_speed(data: SystemBridgeCoordinatorData, key: str) -> float | None:
"""Return the GPU core clock speed."""
if getattr(data.gpu, f"{key}_core_clock") is not None:
return round(getattr(data.gpu, f"{key}_core_clock"))
return None
def gpu_memory_clock_speed(data: SystemBridgeCoordinatorData, key: str) -> float | None:
"""Return the GPU memory clock speed."""
if getattr(data.gpu, f"{key}_memory_clock") is not None:
return round(getattr(data.gpu, f"{key}_memory_clock"))
return None
def gpu_memory_free(data: SystemBridgeCoordinatorData, key: str) -> float | None:
"""Return the free GPU memory."""
if getattr(data.gpu, f"{key}_memory_free") is not None:
return round(getattr(data.gpu, f"{key}_memory_free") / 10**3, 2)
return None
def gpu_memory_used(data: SystemBridgeCoordinatorData, key: str) -> float | None:
"""Return the used GPU memory."""
if getattr(data.gpu, f"{key}_memory_used") is not None:
return round(getattr(data.gpu, f"{key}_memory_used") / 10**3, 2)
return None
def gpu_memory_used_percentage(
data: SystemBridgeCoordinatorData, key: str
) -> float | None:
"""Return the used GPU memory percentage."""
if (
getattr(data.gpu, f"{key}_memory_used") is not None
and getattr(data.gpu, f"{key}_memory_total") is not None
):
return round(
getattr(data.gpu, f"{key}_memory_used")
/ getattr(data.gpu, f"{key}_memory_total")
* 100,
2,
)
return None
def memory_free(data: SystemBridgeCoordinatorData) -> float | None:
"""Return the free memory."""
if data.memory.virtual_free is not None:
return round(data.memory.virtual_free / 1000**3, 2)
return None
def memory_used(data: SystemBridgeCoordinatorData) -> float | None:
"""Return the used memory."""
if data.memory.virtual_used is not None:
return round(data.memory.virtual_used / 1000**3, 2)
return None
BASE_SENSOR_TYPES: tuple[SystemBridgeSensorEntityDescription, ...] = (
SystemBridgeSensorEntityDescription(
key="bios_version",
name="BIOS Version",
entity_registry_enabled_default=False,
icon="mdi:chip",
value=lambda bridge: bridge.system.bios.version,
),
SystemBridgeSensorEntityDescription(
key="cpu_speed",
name="CPU Speed",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=FREQUENCY_GIGAHERTZ,
icon="mdi:speedometer",
value=lambda bridge: bridge.cpu.currentSpeed.avg,
value=cpu_speed,
),
SystemBridgeSensorEntityDescription(
key="cpu_temperature",
@ -73,7 +140,7 @@ BASE_SENSOR_TYPES: tuple[SystemBridgeSensorEntityDescription, ...] = (
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=TEMP_CELSIUS,
value=lambda bridge: bridge.cpu.temperature.main,
value=lambda data: data.cpu.temperature,
),
SystemBridgeSensorEntityDescription(
key="cpu_voltage",
@ -82,21 +149,14 @@ BASE_SENSOR_TYPES: tuple[SystemBridgeSensorEntityDescription, ...] = (
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT,
value=lambda bridge: bridge.cpu.cpu.voltage,
),
SystemBridgeSensorEntityDescription(
key="displays_connected",
name="Displays Connected",
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:monitor",
value=lambda bridge: len(bridge.display.displays),
value=lambda data: data.cpu.voltage,
),
SystemBridgeSensorEntityDescription(
key="kernel",
name="Kernel",
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:devices",
value=lambda bridge: bridge.os.kernel,
value=lambda data: data.system.platform,
),
SystemBridgeSensorEntityDescription(
key="memory_free",
@ -104,7 +164,7 @@ BASE_SENSOR_TYPES: tuple[SystemBridgeSensorEntityDescription, ...] = (
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=DATA_GIGABYTES,
icon="mdi:memory",
value=lambda bridge: round(bridge.memory.free / 1000**3, 2),
value=memory_free,
),
SystemBridgeSensorEntityDescription(
key="memory_used_percentage",
@ -112,7 +172,7 @@ BASE_SENSOR_TYPES: tuple[SystemBridgeSensorEntityDescription, ...] = (
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:memory",
value=lambda bridge: round((bridge.memory.used / bridge.memory.total) * 100, 2),
value=lambda data: data.memory.virtual_percent,
),
SystemBridgeSensorEntityDescription(
key="memory_used",
@ -121,14 +181,14 @@ BASE_SENSOR_TYPES: tuple[SystemBridgeSensorEntityDescription, ...] = (
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=DATA_GIGABYTES,
icon="mdi:memory",
value=lambda bridge: round(bridge.memory.used / 1000**3, 2),
value=memory_used,
),
SystemBridgeSensorEntityDescription(
key="os",
name="Operating System",
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:devices",
value=lambda bridge: f"{bridge.os.distro} {bridge.os.release}",
value=lambda data: f"{data.system.platform} {data.system.platform_version}",
),
SystemBridgeSensorEntityDescription(
key="processes_load",
@ -136,46 +196,19 @@ BASE_SENSOR_TYPES: tuple[SystemBridgeSensorEntityDescription, ...] = (
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
value=lambda bridge: round(bridge.processes.load.currentLoad, 2),
),
SystemBridgeSensorEntityDescription(
key="processes_load_idle",
name="Idle Load",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
value=lambda bridge: round(bridge.processes.load.currentLoadIdle, 2),
),
SystemBridgeSensorEntityDescription(
key="processes_load_system",
name="System Load",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
value=lambda bridge: round(bridge.processes.load.currentLoadSystem, 2),
),
SystemBridgeSensorEntityDescription(
key="processes_load_user",
name="User Load",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
value=lambda bridge: round(bridge.processes.load.currentLoadUser, 2),
value=lambda data: data.cpu.usage,
),
SystemBridgeSensorEntityDescription(
key="version",
name="Version",
icon="mdi:counter",
value=lambda bridge: bridge.information.version,
value=lambda data: data.system.version,
),
SystemBridgeSensorEntityDescription(
key="version_latest",
name="Latest Version",
icon="mdi:counter",
value=lambda bridge: bridge.information.updates.version.new,
value=lambda data: data.system.version_latest,
),
)
@ -186,238 +219,270 @@ BATTERY_SENSOR_TYPES: tuple[SystemBridgeSensorEntityDescription, ...] = (
device_class=SensorDeviceClass.BATTERY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
value=lambda bridge: bridge.battery.percent,
value=lambda data: data.battery.percentage,
),
SystemBridgeSensorEntityDescription(
key="battery_time_remaining",
name="Battery Time Remaining",
device_class=SensorDeviceClass.TIMESTAMP,
state_class=SensorStateClass.MEASUREMENT,
value=lambda bridge: str(
datetime.now() + timedelta(minutes=bridge.battery.timeRemaining)
),
value=battery_time_remaining,
),
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up System Bridge sensor based on a config entry."""
coordinator: SystemBridgeDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
entities = []
for description in BASE_SENSOR_TYPES:
entities.append(SystemBridgeSensor(coordinator, description))
entities.append(
SystemBridgeSensor(coordinator, description, entry.data[CONF_PORT])
)
for key, _ in coordinator.data.filesystem.fsSize.items():
uid = key.replace(":", "")
for partition in coordinator.data.disk.partitions:
entities.append(
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"filesystem_{uid}",
name=f"{key} Space Used",
key=f"filesystem_{partition.replace(':', '')}",
name=f"{partition} Space Used",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:harddisk",
value=lambda bridge, i=key: round(
bridge.filesystem.fsSize[i]["use"], 2
value=lambda data, p=partition: getattr(
data.disk, f"usage_{p}_percent"
),
),
entry.data[CONF_PORT],
)
)
if coordinator.data.battery.hasBattery:
if (
coordinator.data.battery
and coordinator.data.battery.percentage
and coordinator.data.battery.percentage > -1
):
for description in BATTERY_SENSOR_TYPES:
entities.append(SystemBridgeSensor(coordinator, description))
entities.append(
SystemBridgeSensor(coordinator, description, entry.data[CONF_PORT])
)
for index, _ in enumerate(coordinator.data.display.displays):
name = index + 1
displays = []
for display in coordinator.data.display.displays:
displays.append(
{
"key": display,
"name": getattr(coordinator.data.display, f"{display}_name").replace(
"Display ", ""
),
},
)
display_count = len(displays)
entities.append(
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key="displays_connected",
name="Displays Connected",
state_class=SensorStateClass.MEASUREMENT,
icon="mdi:monitor",
value=lambda _, count=display_count: count,
),
entry.data[CONF_PORT],
)
)
for _, display in enumerate(displays):
entities = [
*entities,
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"display_{name}_resolution_x",
name=f"Display {name} Resolution X",
key=f"display_{display['name']}_resolution_x",
name=f"Display {display['name']} Resolution X",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PIXELS,
icon="mdi:monitor",
value=lambda bridge, i=index: bridge.display.displays[
i
].resolutionX,
value=lambda data, k=display["key"]: getattr(
data.display, f"{k}_resolution_horizontal"
),
),
entry.data[CONF_PORT],
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"display_{name}_resolution_y",
name=f"Display {name} Resolution Y",
key=f"display_{display['name']}_resolution_y",
name=f"Display {display['name']} Resolution Y",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PIXELS,
icon="mdi:monitor",
value=lambda bridge, i=index: bridge.display.displays[
i
].resolutionY,
value=lambda data, k=display["key"]: getattr(
data.display, f"{k}_resolution_vertical"
),
),
entry.data[CONF_PORT],
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"display_{name}_refresh_rate",
name=f"Display {name} Refresh Rate",
key=f"display_{display['name']}_refresh_rate",
name=f"Display {display['name']} Refresh Rate",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=FREQUENCY_HERTZ,
icon="mdi:monitor",
value=lambda bridge, i=index: bridge.display.displays[
i
].currentRefreshRate,
value=lambda data, k=display["key"]: getattr(
data.display, f"{k}_refresh_rate"
),
),
entry.data[CONF_PORT],
),
]
for index, _ in enumerate(coordinator.data.graphics.controllers):
if coordinator.data.graphics.controllers[index].name is not None:
# Remove vendor from name
name = (
coordinator.data.graphics.controllers[index]
.name.replace(coordinator.data.graphics.controllers[index].vendor, "")
.strip()
)
entities = [
*entities,
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_core_clock_speed",
name=f"{name} Clock Speed",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=FREQUENCY_MEGAHERTZ,
icon="mdi:speedometer",
value=lambda bridge, i=index: bridge.graphics.controllers[
i
].clockCore,
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_memory_clock_speed",
name=f"{name} Memory Clock Speed",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=FREQUENCY_MEGAHERTZ,
icon="mdi:speedometer",
value=lambda bridge, i=index: bridge.graphics.controllers[
i
].clockMemory,
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_memory_free",
name=f"{name} Memory Free",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=DATA_GIGABYTES,
icon="mdi:memory",
value=lambda bridge, i=index: round(
bridge.graphics.controllers[i].memoryFree / 10**3, 2
),
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_memory_used_percentage",
name=f"{name} Memory Used %",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:memory",
value=lambda bridge, i=index: round(
(
bridge.graphics.controllers[i].memoryUsed
/ bridge.graphics.controllers[i].memoryTotal
)
* 100,
2,
),
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_memory_used",
name=f"{name} Memory Used",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=DATA_GIGABYTES,
icon="mdi:memory",
value=lambda bridge, i=index: round(
bridge.graphics.controllers[i].memoryUsed / 10**3, 2
),
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_fan_speed",
name=f"{name} Fan Speed",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:fan",
value=lambda bridge, i=index: bridge.graphics.controllers[
i
].fanSpeed,
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_power_usage",
name=f"{name} Power Usage",
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=POWER_WATT,
value=lambda bridge, i=index: bridge.graphics.controllers[
i
].powerDraw,
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_temperature",
name=f"{name} Temperature",
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=TEMP_CELSIUS,
value=lambda bridge, i=index: bridge.graphics.controllers[
i
].temperatureGpu,
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_usage_percentage",
name=f"{name} Usage %",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
value=lambda bridge, i=index: bridge.graphics.controllers[
i
].utilizationGpu,
),
),
]
gpus = []
for gpu in coordinator.data.gpu.gpus:
gpus.append(
{
"key": gpu,
"name": getattr(coordinator.data.gpu, f"{gpu}_name"),
},
)
for index, _ in enumerate(coordinator.data.processes.load.cpus):
for index, gpu in enumerate(gpus):
entities = [
*entities,
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_core_clock_speed",
name=f"{gpu['name']} Clock Speed",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=FREQUENCY_MEGAHERTZ,
icon="mdi:speedometer",
value=lambda data, k=gpu["key"]: gpu_core_clock_speed(data, k),
),
entry.data[CONF_PORT],
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_memory_clock_speed",
name=f"{gpu['name']} Memory Clock Speed",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=FREQUENCY_MEGAHERTZ,
icon="mdi:speedometer",
value=lambda data, k=gpu["key"]: gpu_memory_clock_speed(data, k),
),
entry.data[CONF_PORT],
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_memory_free",
name=f"{gpu['name']} Memory Free",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=DATA_GIGABYTES,
icon="mdi:memory",
value=lambda data, k=gpu["key"]: gpu_memory_free(data, k),
),
entry.data[CONF_PORT],
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_memory_used_percentage",
name=f"{gpu['name']} Memory Used %",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:memory",
value=lambda data, k=gpu["key"]: gpu_memory_used_percentage(
data, k
),
),
entry.data[CONF_PORT],
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_memory_used",
name=f"{gpu['name']} Memory Used",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=DATA_GIGABYTES,
icon="mdi:memory",
value=lambda data, k=gpu["key"]: gpu_memory_used(data, k),
),
entry.data[CONF_PORT],
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_fan_speed",
name=f"{gpu['name']} Fan Speed",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=RPM,
icon="mdi:fan",
value=lambda data, k=gpu["key"]: getattr(
data.gpu, f"{k}_fan_speed"
),
),
entry.data[CONF_PORT],
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_power_usage",
name=f"{gpu['name']} Power Usage",
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=POWER_WATT,
value=lambda data, k=gpu["key"]: getattr(data.gpu, f"{k}_power"),
),
entry.data[CONF_PORT],
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_temperature",
name=f"{gpu['name']} Temperature",
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=TEMP_CELSIUS,
value=lambda data, k=gpu["key"]: getattr(
data.gpu, f"{k}_temperature"
),
),
entry.data[CONF_PORT],
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"gpu_{index}_usage_percentage",
name=f"{gpu['name']} Usage %",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
value=lambda data, k=gpu["key"]: getattr(
data.gpu, f"{k}_core_load"
),
),
entry.data[CONF_PORT],
),
]
for index in range(coordinator.data.cpu.count):
entities = [
*entities,
SystemBridgeSensor(
@ -429,52 +494,9 @@ async def async_setup_entry(
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
value=lambda bridge, index=index: round(
bridge.processes.load.cpus[index].load, 2
),
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"processes_load_cpu_{index}_idle",
name=f"Idle Load CPU {index}",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
value=lambda bridge, index=index: round(
bridge.processes.load.cpus[index].loadIdle, 2
),
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"processes_load_cpu_{index}_system",
name=f"System Load CPU {index}",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
value=lambda bridge, index=index: round(
bridge.processes.load.cpus[index].loadSystem, 2
),
),
),
SystemBridgeSensor(
coordinator,
SystemBridgeSensorEntityDescription(
key=f"processes_load_cpu_{index}_user",
name=f"User Load CPU {index}",
entity_registry_enabled_default=False,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
icon="mdi:percent",
value=lambda bridge, index=index: round(
bridge.processes.load.cpus[index].loadUser, 2
),
value=lambda data, k=index: getattr(data.cpu, f"usage_{k}"),
),
entry.data[CONF_PORT],
),
]
@ -490,10 +512,12 @@ class SystemBridgeSensor(SystemBridgeDeviceEntity, SensorEntity):
self,
coordinator: SystemBridgeDataUpdateCoordinator,
description: SystemBridgeSensorEntityDescription,
api_port: int,
) -> None:
"""Initialize."""
super().__init__(
coordinator,
api_port,
description.key,
description.name,
)
@ -502,8 +526,7 @@ class SystemBridgeSensor(SystemBridgeDeviceEntity, SensorEntity):
@property
def native_value(self) -> StateType:
"""Return the state."""
bridge: Bridge = self.coordinator.data
try:
return cast(StateType, self.entity_description.value(bridge))
return cast(StateType, self.entity_description.value(self.coordinator.data))
except TypeError:
return None

64
homeassistant/components/system_bridge/services.yaml
View file

@ -1,32 +1,6 @@
send_command:
name: Send Command
description: Sends a command to the server to run.
fields:
bridge:
name: Bridge
description: The server to send the command to.
required: true
selector:
device:
integration: system_bridge
command:
name: Command
description: Command to send to the server.
required: true
example: "echo"
selector:
text:
arguments:
name: Arguments
description: Arguments to send to the server.
required: false
default: ""
example: "hello"
selector:
text:
open:
name: Open Path/URL
description: Open a URL or file on the server using the default application.
open_path:
name: Open Path
description: Open a file on the server using the default application.
fields:
bridge:
name: Bridge
@ -36,8 +10,26 @@ open:
device:
integration: system_bridge
path:
name: Path/URL
description: Path/URL to open.
name: Path
description: Path to open.
required: true
example: "C:\\test\\image.png"
selector:
text:
open_url:
name: Open URL
description: Open a URL on the server using the default application.
fields:
bridge:
name: Bridge
description: The server to talk to.
required: true
selector:
device:
integration: system_bridge
url:
name: URL
description: URL to open.
required: true
example: "https://www.home-assistant.io"
selector:
@ -60,16 +52,6 @@ send_keypress:
example: "audio_play"
selector:
text:
modifiers:
name: Modifiers
description: "List of modifier(s). Accepts alt, command/win, control, and shift."
required: false
default: ""
example:
- "control"
- "shift"
selector:
text:
send_text:
name: Send Keyboard Text
description: Sends text for the server to type.

2
homeassistant/generated/zeroconf.py
View file

@ -370,7 +370,7 @@ ZEROCONF = {
"name": "smappee50*"
}
],
"_system-bridge._udp.local.": [
"_system-bridge._tcp.local.": [
{
"domain": "system_bridge"
}

2
requirements_all.txt
View file

@ -2271,7 +2271,7 @@ swisshydrodata==0.1.0
synology-srm==0.2.0
# homeassistant.components.system_bridge
systembridge==2.3.1
systembridgeconnector==3.1.3
# homeassistant.components.tailscale
tailscale==0.2.0

2
requirements_test_all.txt
View file

@ -1498,7 +1498,7 @@ sunwatcher==0.2.1
surepy==0.7.2
# homeassistant.components.system_bridge
systembridge==2.3.1
systembridgeconnector==3.1.3
# homeassistant.components.tailscale
tailscale==0.2.0

365
tests/components/system_bridge/test_config_flow.py
View file

@ -1,13 +1,28 @@
"""Test the System Bridge config flow."""
import asyncio
from unittest.mock import patch
from aiohttp.client_exceptions import ClientConnectionError
from systembridge.exceptions import BridgeAuthenticationException
from systembridgeconnector.const import (
EVENT_DATA,
EVENT_MESSAGE,
EVENT_MODULE,
EVENT_SUBTYPE,
EVENT_TYPE,
SUBTYPE_BAD_API_KEY,
TYPE_DATA_UPDATE,
TYPE_ERROR,
)
from systembridgeconnector.exceptions import (
AuthenticationException,
ConnectionClosedException,
ConnectionErrorException,
)
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import zeroconf
from homeassistant.components.system_bridge.const import DOMAIN
from homeassistant.const import CONF_API_KEY, CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@ -29,7 +44,7 @@ FIXTURE_ZEROCONF_INPUT = {
}
FIXTURE_ZEROCONF = zeroconf.ZeroconfServiceInfo(
host="1.1.1.1",
host="test-bridge",
addresses=["1.1.1.1"],
port=9170,
hostname="test-bridge.local.",
@ -58,37 +73,40 @@ FIXTURE_ZEROCONF_BAD = zeroconf.ZeroconfServiceInfo(
},
)
FIXTURE_INFORMATION = {
"address": "http://test-bridge:9170",
"apiPort": 9170,
"fqdn": "test-bridge",
"host": "test-bridge",
"ip": "1.1.1.1",
"mac": FIXTURE_MAC_ADDRESS,
"updates": {
"available": False,
"newer": False,
"url": "https://github.com/timmo001/system-bridge/releases/tag/v2.3.2",
"version": {"current": "2.3.2", "new": "2.3.2"},
FIXTURE_DATA_SYSTEM = {
EVENT_TYPE: TYPE_DATA_UPDATE,
EVENT_MESSAGE: "Data changed",
EVENT_MODULE: "system",
EVENT_DATA: {
"uuid": FIXTURE_UUID,
},
"uuid": FIXTURE_UUID,
"version": "2.3.2",
"websocketAddress": "ws://test-bridge:9172",
"websocketPort": 9172,
}
FIXTURE_DATA_SYSTEM_BAD = {
EVENT_TYPE: TYPE_DATA_UPDATE,
EVENT_MESSAGE: "Data changed",
EVENT_MODULE: "system",
EVENT_DATA: {},
}
FIXTURE_DATA_AUTH_ERROR = {
EVENT_TYPE: TYPE_ERROR,
EVENT_SUBTYPE: SUBTYPE_BAD_API_KEY,
EVENT_MESSAGE: "Invalid api-key",
}
FIXTURE_BASE_URL = (
f"http://{FIXTURE_USER_INPUT[CONF_HOST]}:{FIXTURE_USER_INPUT[CONF_PORT]}"
)
async def test_show_user_form(hass: HomeAssistant) -> None:
"""Test that the setup form is served."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
FIXTURE_ZEROCONF_BASE_URL = f"http://{FIXTURE_ZEROCONF.host}:{FIXTURE_ZEROCONF.port}"
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
async def test_user_flow(
hass, aiohttp_client, aioclient_mock, current_request_with_host
) -> None:
async def test_user_flow(hass: HomeAssistant) -> None:
"""Test full user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -97,20 +115,19 @@ async def test_user_flow(
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] is None
aioclient_mock.get(
f"{FIXTURE_BASE_URL}/information",
headers={"Content-Type": "application/json"},
json=FIXTURE_INFORMATION,
)
with patch(
"homeassistant.components.system_bridge.config_flow.WebSocketClient.connect"
), patch("systembridgeconnector.websocket_client.WebSocketClient.get_data"), patch(
"systembridgeconnector.websocket_client.WebSocketClient.receive_message",
return_value=FIXTURE_DATA_SYSTEM,
), patch(
"homeassistant.components.system_bridge.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT
)
await hass.async_block_till_done()
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "test-bridge"
@ -118,34 +135,7 @@ async def test_user_flow(
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_invalid_auth(
hass, aiohttp_client, aioclient_mock, current_request_with_host
) -> None:
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] is None
aioclient_mock.get(
f"{FIXTURE_BASE_URL}/information", exc=BridgeAuthenticationException
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "invalid_auth"}
async def test_form_cannot_connect(
hass, aiohttp_client, aioclient_mock, current_request_with_host
) -> None:
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -154,11 +144,13 @@ async def test_form_cannot_connect(
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] is None
aioclient_mock.get(f"{FIXTURE_BASE_URL}/information", exc=ClientConnectionError)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT
)
with patch(
"systembridgeconnector.websocket_client.WebSocketClient.connect",
side_effect=ConnectionErrorException,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
@ -166,10 +158,8 @@ async def test_form_cannot_connect(
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_unknown_error(
hass, aiohttp_client, aioclient_mock, current_request_with_host
) -> None:
"""Test we handle unknown error."""
async def test_form_connection_closed_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle connection closed cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
@ -177,23 +167,123 @@ async def test_form_unknown_error(
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] is None
with patch(
"homeassistant.components.system_bridge.config_flow.Bridge.async_get_information",
side_effect=Exception("Boom"),
with patch("systembridgeconnector.websocket_client.WebSocketClient.connect"), patch(
"systembridgeconnector.websocket_client.WebSocketClient.get_data"
), patch(
"systembridgeconnector.websocket_client.WebSocketClient.receive_message",
side_effect=ConnectionClosedException,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT
)
await hass.async_block_till_done()
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_timeout_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle timeout cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] is None
with patch("systembridgeconnector.websocket_client.WebSocketClient.connect"), patch(
"systembridgeconnector.websocket_client.WebSocketClient.get_data"
), patch(
"systembridgeconnector.websocket_client.WebSocketClient.receive_message",
side_effect=asyncio.TimeoutError,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_invalid_auth(hass: HomeAssistant) -> None:
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] is None
with patch("systembridgeconnector.websocket_client.WebSocketClient.connect"), patch(
"systembridgeconnector.websocket_client.WebSocketClient.get_data"
), patch(
"systembridgeconnector.websocket_client.WebSocketClient.receive_message",
side_effect=AuthenticationException,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "invalid_auth"}
async def test_form_uuid_error(hass: HomeAssistant) -> None:
"""Test we handle error from bad uuid."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] is None
with patch("systembridgeconnector.websocket_client.WebSocketClient.connect"), patch(
"systembridgeconnector.websocket_client.WebSocketClient.get_data"
), patch(
"systembridgeconnector.websocket_client.WebSocketClient.receive_message",
return_value=FIXTURE_DATA_SYSTEM_BAD,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_unknown_error(hass: HomeAssistant) -> None:
"""Test we handle unknown errors."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] is None
with patch("systembridgeconnector.websocket_client.WebSocketClient.connect"), patch(
"systembridgeconnector.websocket_client.WebSocketClient.get_data"
), patch(
"systembridgeconnector.websocket_client.WebSocketClient.receive_message",
side_effect=Exception,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_USER_INPUT
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["step_id"] == "user"
assert result2["errors"] == {"base": "unknown"}
async def test_reauth_authorization_error(
hass, aiohttp_client, aioclient_mock, current_request_with_host
) -> None:
async def test_reauth_authorization_error(hass: HomeAssistant) -> None:
"""Test we show user form on authorization error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "reauth"}, data=FIXTURE_USER_INPUT
@ -202,13 +292,15 @@ async def test_reauth_authorization_error(
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "authenticate"
aioclient_mock.get(
f"{FIXTURE_BASE_URL}/information", exc=BridgeAuthenticationException
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_AUTH_INPUT
)
with patch("systembridgeconnector.websocket_client.WebSocketClient.connect"), patch(
"systembridgeconnector.websocket_client.WebSocketClient.get_data"
), patch(
"systembridgeconnector.websocket_client.WebSocketClient.receive_message",
side_effect=AuthenticationException,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_AUTH_INPUT
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
@ -216,9 +308,7 @@ async def test_reauth_authorization_error(
assert result2["errors"] == {"base": "invalid_auth"}
async def test_reauth_connection_error(
hass, aiohttp_client, aioclient_mock, current_request_with_host
) -> None:
async def test_reauth_connection_error(hass: HomeAssistant) -> None:
"""Test we show user form on connection error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "reauth"}, data=FIXTURE_USER_INPUT
@ -227,11 +317,13 @@ async def test_reauth_connection_error(
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "authenticate"
aioclient_mock.get(f"{FIXTURE_BASE_URL}/information", exc=ClientConnectionError)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_AUTH_INPUT
)
with patch(
"systembridgeconnector.websocket_client.WebSocketClient.connect",
side_effect=ConnectionErrorException,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_AUTH_INPUT
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
@ -239,9 +331,32 @@ async def test_reauth_connection_error(
assert result2["errors"] == {"base": "cannot_connect"}
async def test_reauth_flow(
hass, aiohttp_client, aioclient_mock, current_request_with_host
) -> None:
async def test_reauth_connection_closed_error(hass: HomeAssistant) -> None:
"""Test we show user form on connection error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "reauth"}, data=FIXTURE_USER_INPUT
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "authenticate"
with patch("systembridgeconnector.websocket_client.WebSocketClient.connect"), patch(
"systembridgeconnector.websocket_client.WebSocketClient.get_data"
), patch(
"systembridgeconnector.websocket_client.WebSocketClient.receive_message",
side_effect=ConnectionClosedException,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_AUTH_INPUT
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result2["step_id"] == "authenticate"
assert result2["errors"] == {"base": "cannot_connect"}
async def test_reauth_flow(hass: HomeAssistant) -> None:
"""Test reauth flow."""
mock_config = MockConfigEntry(
domain=DOMAIN, unique_id=FIXTURE_UUID, data=FIXTURE_USER_INPUT
@ -255,20 +370,19 @@ async def test_reauth_flow(
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "authenticate"
aioclient_mock.get(
f"{FIXTURE_BASE_URL}/information",
headers={"Content-Type": "application/json"},
json=FIXTURE_INFORMATION,
)
with patch(
with patch("systembridgeconnector.websocket_client.WebSocketClient.connect"), patch(
"systembridgeconnector.websocket_client.WebSocketClient.get_data"
), patch(
"systembridgeconnector.websocket_client.WebSocketClient.receive_message",
return_value=FIXTURE_DATA_SYSTEM,
), patch(
"homeassistant.components.system_bridge.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_AUTH_INPUT
)
await hass.async_block_till_done()
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result2["reason"] == "reauth_successful"
@ -276,9 +390,7 @@ async def test_reauth_flow(
assert len(mock_setup_entry.mock_calls) == 1
async def test_zeroconf_flow(
hass, aiohttp_client, aioclient_mock, current_request_with_host
) -> None:
async def test_zeroconf_flow(hass: HomeAssistant) -> None:
"""Test zeroconf flow."""
result = await hass.config_entries.flow.async_init(
@ -290,30 +402,27 @@ async def test_zeroconf_flow(
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert not result["errors"]
aioclient_mock.get(
f"{FIXTURE_ZEROCONF_BASE_URL}/information",
headers={"Content-Type": "application/json"},
json=FIXTURE_INFORMATION,
)
with patch(
with patch("systembridgeconnector.websocket_client.WebSocketClient.connect"), patch(
"systembridgeconnector.websocket_client.WebSocketClient.get_data"
), patch(
"systembridgeconnector.websocket_client.WebSocketClient.receive_message",
return_value=FIXTURE_DATA_SYSTEM,
), patch(
"homeassistant.components.system_bridge.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_AUTH_INPUT
)
await hass.async_block_till_done()
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == "test-bridge"
assert result2["title"] == "1.1.1.1"
assert result2["data"] == FIXTURE_ZEROCONF_INPUT
assert len(mock_setup_entry.mock_calls) == 1
async def test_zeroconf_cannot_connect(
hass, aiohttp_client, aioclient_mock, current_request_with_host
) -> None:
async def test_zeroconf_cannot_connect(hass: HomeAssistant) -> None:
"""Test zeroconf cannot connect flow."""
result = await hass.config_entries.flow.async_init(
@ -325,13 +434,13 @@ async def test_zeroconf_cannot_connect(
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert not result["errors"]
aioclient_mock.get(
f"{FIXTURE_ZEROCONF_BASE_URL}/information", exc=ClientConnectionError
)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_AUTH_INPUT
)
with patch(
"systembridgeconnector.websocket_client.WebSocketClient.connect",
side_effect=ConnectionErrorException,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], FIXTURE_AUTH_INPUT
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_FORM
@ -339,9 +448,7 @@ async def test_zeroconf_cannot_connect(
assert result2["errors"] == {"base": "cannot_connect"}
async def test_zeroconf_bad_zeroconf_info(
hass, aiohttp_client, aioclient_mock, current_request_with_host
) -> None:
async def test_zeroconf_bad_zeroconf_info(hass: HomeAssistant) -> None:
"""Test zeroconf cannot connect flow."""
result = await hass.config_entries.flow.async_init(