jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 09:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Sensoterra integration (#119642)

Browse Source 
* Initial version

* Baseline release

* Refactor based on first PR feedback

* Refactoring based on second PR feedback

* Initial version

* Baseline release

* Refactor based on first PR feedback

* Refactoring based on second PR feedback

* Refactoring based on PR feedback

* Refactoring based on PR feedback

* Remove extra attribute soil type

Soil type isn't really a sensor, but more like a configuration entity.
Move soil type to a different PR to keep this PR simpler.

* Refactor SensoterraSensor to a named tuple

* Implement feedback on PR

* Remove .coveragerc

* Add async_set_unique_id to config flow

* Small fix based on feedback

* Add test form unique_id

* Fix

* Fix

---------

Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Mark Ruys 2024-09-05 21:37:44 +02:00 committed by GitHub
parent bbeecb40ae
commit 9e312f2063
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 601 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -401,6 +401,7 @@ homeassistant.components.select.*
homeassistant.components.sensibo.*
homeassistant.components.sensirion_ble.*
homeassistant.components.sensor.*
homeassistant.components.sensoterra.*
homeassistant.components.senz.*
homeassistant.components.sfr_box.*
homeassistant.components.shelly.*

2
CODEOWNERS
View File

@ -1288,6 +1288,8 @@ build.json @home-assistant/supervisor
/tests/components/sensorpro/ @bdraco
/homeassistant/components/sensorpush/ @bdraco
/tests/components/sensorpush/ @bdraco
/homeassistant/components/sensoterra/ @markruys
/tests/components/sensoterra/ @markruys
/homeassistant/components/sentry/ @dcramer @frenck
/tests/components/sentry/ @dcramer @frenck
/homeassistant/components/senz/ @milanmeu

38
homeassistant/components/sensoterra/__init__.py Normal file
View File

@ -0,0 +1,38 @@
"""The Sensoterra integration."""
from __future__ import annotations
from sensoterra.customerapi import CustomerApi
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_TOKEN, Platform
from homeassistant.core import HomeAssistant
from .coordinator import SensoterraCoordinator
PLATFORMS: list[Platform] = [Platform.SENSOR]
type SensoterraConfigEntry = ConfigEntry[SensoterraCoordinator]
async def async_setup_entry(hass: HomeAssistant, entry: SensoterraConfigEntry) -> bool:
"""Set up Sensoterra platform based on a configuration entry."""
# Create a coordinator and add an API instance to it. Store the coordinator
# in the configuration entry.
api = CustomerApi()
api.set_language(hass.config.language)
api.set_token(entry.data[CONF_TOKEN])
coordinator = SensoterraCoordinator(hass, api)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: SensoterraConfigEntry) -> bool:
"""Unload the configuration entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

90
homeassistant/components/sensoterra/config_flow.py Normal file
View File

@ -0,0 +1,90 @@
"""Config flow for Sensoterra integration."""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import Any
from jwt import DecodeError, decode
from sensoterra.customerapi import (
CustomerApi,
InvalidAuth as StInvalidAuth,
Timeout as StTimeout,
)
import voluptuous as vol
from homeassistant.config_entries import SOURCE_USER, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, CONF_TOKEN
from homeassistant.helpers.selector import (
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from .const import DOMAIN, LOGGER, TOKEN_EXPIRATION_DAYS
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_EMAIL): TextSelector(
TextSelectorConfig(type=TextSelectorType.EMAIL, autocomplete="email")
),
vol.Required(CONF_PASSWORD): TextSelector(
TextSelectorConfig(type=TextSelectorType.PASSWORD)
),
}
)
class SensoterraConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Sensoterra."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Create hub entry based on config flow."""
errors: dict[str, str] = {}
if user_input is not None:
api = CustomerApi(user_input[CONF_EMAIL], user_input[CONF_PASSWORD])
# We need a unique tag per HA instance
uuid = self.hass.data["core.uuid"]
expiration = datetime.now() + timedelta(TOKEN_EXPIRATION_DAYS)
try:
token: str = await api.get_token(
f"Home Assistant {uuid}", "READONLY", expiration
)
decoded_token = decode(
token, algorithms=["HS256"], options={"verify_signature": False}
)
except StInvalidAuth as exp:
LOGGER.error(
"Login attempt with %s: %s", user_input[CONF_EMAIL], exp.message
)
errors["base"] = "invalid_auth"
except StTimeout:
LOGGER.error("Login attempt with %s: time out", user_input[CONF_EMAIL])
errors["base"] = "cannot_connect"
except DecodeError:
LOGGER.error("Login attempt with %s: bad token", user_input[CONF_EMAIL])
errors["base"] = "invalid_access_token"
else:
device_unique_id = decoded_token["sub"]
await self.async_set_unique_id(device_unique_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input[CONF_EMAIL],
data={
CONF_TOKEN: token,
CONF_EMAIL: user_input[CONF_EMAIL],
},
)
return self.async_show_form(
step_id=SOURCE_USER,
data_schema=self.add_suggested_values_to_schema(
STEP_USER_DATA_SCHEMA, user_input
),
errors=errors,
)

10
homeassistant/components/sensoterra/const.py Normal file
View File

@ -0,0 +1,10 @@
"""Constants for the Sensoterra integration."""
import logging
DOMAIN = "sensoterra"
SCAN_INTERVAL_MINUTES = 15
SENSOR_EXPIRATION_DAYS = 2
TOKEN_EXPIRATION_DAYS = 10 * 365
CONFIGURATION_URL = "https://monitor.sensoterra.com"
LOGGER: logging.Logger = logging.getLogger(__package__)

54
homeassistant/components/sensoterra/coordinator.py Normal file
View File

@ -0,0 +1,54 @@
"""Polling coordinator for the Sensoterra integration."""
from collections.abc import Callable
from datetime import timedelta
from sensoterra.customerapi import (
CustomerApi,
InvalidAuth as ApiAuthError,
Timeout as ApiTimeout,
)
from sensoterra.probe import Probe, Sensor
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import LOGGER, SCAN_INTERVAL_MINUTES
class SensoterraCoordinator(DataUpdateCoordinator[list[Probe]]):
"""Sensoterra coordinator."""
def __init__(self, hass: HomeAssistant, api: CustomerApi) -> None:
"""Initialize Sensoterra coordinator."""
super().__init__(
hass,
LOGGER,
name="Sensoterra probe",
update_interval=timedelta(minutes=SCAN_INTERVAL_MINUTES),
)
self.api = api
self.add_devices_callback: Callable[[list[Probe]], None] | None = None
async def _async_update_data(self) -> list[Probe]:
"""Fetch data from Sensoterra Customer API endpoint."""
try:
probes = await self.api.poll()
except ApiAuthError as err:
raise ConfigEntryError(err) from err
except ApiTimeout as err:
raise UpdateFailed("Timeout communicating with Sensotera API") from err
if self.add_devices_callback is not None:
self.add_devices_callback(probes)
return probes
def get_sensor(self, id: str | None) -> Sensor | None:
"""Try to find the sensor in the API result."""
for probe in self.data:
for sensor in probe.sensors():
if sensor.id == id:
return sensor
return None

10
homeassistant/components/sensoterra/manifest.json Normal file
View File

@ -0,0 +1,10 @@
{
"domain": "sensoterra",
"name": "Sensoterra",
"codeowners": ["@markruys"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/sensoterra",
"integration_type": "hub",
"iot_class": "cloud_polling",
"requirements": ["sensoterra==2.0.1"]
}

172
homeassistant/components/sensoterra/sensor.py Normal file
View File

@ -0,0 +1,172 @@
"""Sensoterra devices."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from enum import StrEnum, auto
from sensoterra.probe import Probe, Sensor
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
EntityCategory,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import SensoterraConfigEntry
from .const import CONFIGURATION_URL, DOMAIN, SENSOR_EXPIRATION_DAYS
from .coordinator import SensoterraCoordinator
class ProbeSensorType(StrEnum):
"""Generic sensors within a Sensoterra probe."""
MOISTURE = auto()
SI = auto()
TEMPERATURE = auto()
BATTERY = auto()
RSSI = auto()
SENSORS: dict[ProbeSensorType, SensorEntityDescription] = {
ProbeSensorType.MOISTURE: SensorEntityDescription(
key=ProbeSensorType.MOISTURE,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=0,
device_class=SensorDeviceClass.MOISTURE,
native_unit_of_measurement=PERCENTAGE,
translation_key="soil_moisture_at_cm",
),
ProbeSensorType.SI: SensorEntityDescription(
key=ProbeSensorType.SI,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=1,
translation_key="si_at_cm",
),
ProbeSensorType.TEMPERATURE: SensorEntityDescription(
key=ProbeSensorType.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=0,
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
),
ProbeSensorType.BATTERY: SensorEntityDescription(
key=ProbeSensorType.BATTERY,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=0,
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
),
ProbeSensorType.RSSI: SensorEntityDescription(
key=ProbeSensorType.RSSI,
state_class=SensorStateClass.MEASUREMENT,
suggested_display_precision=0,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
}
async def async_setup_entry(
hass: HomeAssistant,
entry: SensoterraConfigEntry,
async_add_devices: AddEntitiesCallback,
) -> None:
"""Set up Sensoterra sensor."""
coordinator = entry.runtime_data
@callback
def _async_add_devices(probes: list[Probe]) -> None:
aha = coordinator.async_contexts()
current_sensors = set(aha)
async_add_devices(
SensoterraEntity(
coordinator,
probe,
sensor,
SENSORS[ProbeSensorType[sensor.type]],
)
for probe in probes
for sensor in probe.sensors()
if sensor.type is not None
and sensor.type.lower() in SENSORS
and sensor.id not in current_sensors
)
coordinator.add_devices_callback = _async_add_devices
_async_add_devices(coordinator.data)
class SensoterraEntity(CoordinatorEntity[SensoterraCoordinator], SensorEntity):
"""Sensoterra sensor like a soil moisture or temperature sensor."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: SensoterraCoordinator,
probe: Probe,
sensor: Sensor,
entity_description: SensorEntityDescription,
) -> None:
"""Initialize entity."""
super().__init__(coordinator, context=sensor.id)
self._sensor_id = sensor.id
self._attr_unique_id = self._sensor_id
self._attr_translation_placeholders = {
"depth": "?" if sensor.depth is None else str(sensor.depth)
}
self.entity_description = entity_description
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, probe.serial)},
name=probe.name,
model=probe.sku,
manufacturer="Sensoterra",
serial_number=probe.serial,
suggested_area=probe.location,
configuration_url=CONFIGURATION_URL,
)
@property
def sensor(self) -> Sensor | None:
"""Return the sensor, or None if it doesn't exist."""
return self.coordinator.get_sensor(self._sensor_id)
@property
def native_value(self) -> StateType:
"""Return the value reported by the sensor."""
assert self.sensor
return self.sensor.value
@property
def available(self) -> bool:
"""Return True if entity is available."""
if not super().available or (sensor := self.sensor) is None:
return False
if sensor.timestamp is None:
return False
# Expire sensor if no update within the last few days.
expiration = datetime.now(UTC) - timedelta(days=SENSOR_EXPIRATION_DAYS)
return sensor.timestamp >= expiration

38
homeassistant/components/sensoterra/strings.json Normal file
View File

@ -0,0 +1,38 @@
{
"config": {
"step": {
"user": {
"description": "Enter credentials to obtain a token",
"data": {
"email": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]"
}
},
"reconfigure": {
"description": "[%key:component::sensoterra::config::step::user::description%]",
"data": {
"email": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_access_token": "[%key:common::config_flow::error::invalid_access_token%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
}
},
"entity": {
"sensor": {
"soil_moisture_at_cm": {
"name": "Soil moisture @ {depth} cm"
},
"si_at_cm": {
"name": "SI @ {depth} cm"
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -513,6 +513,7 @@ FLOWS = {
"sensirion_ble",
"sensorpro",
"sensorpush",
"sensoterra",
"sentry",
"senz",
"seventeentrack",

6
homeassistant/generated/integrations.json
View File

@ -5391,6 +5391,12 @@
"config_flow": true,
"iot_class": "local_push"
},
"sensoterra": {
"name": "Sensoterra",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"sentry": {
"name": "Sentry",
"integration_type": "service",

10
mypy.ini
View File

@ -3766,6 +3766,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.sensoterra.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.senz.*]
check_untyped_defs = true
disallow_incomplete_defs = true

3
requirements_all.txt
View File

@ -2609,6 +2609,9 @@ sensorpro-ble==0.5.3
# homeassistant.components.sensorpush
sensorpush-ble==1.6.2
# homeassistant.components.sensoterra
sensoterra==2.0.1
# homeassistant.components.sentry
sentry-sdk==1.40.3

3
requirements_test_all.txt
View File

@ -2067,6 +2067,9 @@ sensorpro-ble==0.5.3
# homeassistant.components.sensorpush
sensorpush-ble==1.6.2
# homeassistant.components.sensoterra
sensoterra==2.0.1
# homeassistant.components.sentry
sentry-sdk==1.40.3

1
tests/components/sensoterra/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Sensoterra integration."""

32
tests/components/sensoterra/conftest.py Normal file
View File

@ -0,0 +1,32 @@
"""Common fixtures for the Sensoterra tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
from .const import API_TOKEN
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.sensoterra.async_setup_entry",
return_value=True,
) as mock_entry:
yield mock_entry
@pytest.fixture
def mock_customer_api_client() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with (
patch(
"homeassistant.components.sensoterra.config_flow.CustomerApi",
autospec=True,
) as mock_client,
):
mock = mock_client.return_value
mock.get_token.return_value = API_TOKEN
yield mock

7
tests/components/sensoterra/const.py Normal file
View File

@ -0,0 +1,7 @@
"""Constants for the test Sensoterra integration."""
API_TOKEN = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE4NTYzMDQwMDAsInN1YiI6IjM5In0.yxdXXlc1DqopqDRHfAVzFrMqZJl6nKLpu1dV8alHvVY"
API_EMAIL = "test-email@example.com"
API_PASSWORD = "test-password"
HASS_UUID = "phony-unique-id"
SOURCE_USER = "user"

123
tests/components/sensoterra/test_config_flow.py Normal file
View File

@ -0,0 +1,123 @@
"""Test the Sensoterra config flow."""
from unittest.mock import AsyncMock
from jwt import DecodeError
import pytest
from sensoterra.customerapi import InvalidAuth as StInvalidAuth, Timeout as StTimeout
from homeassistant.components.sensoterra.const import DOMAIN
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD, CONF_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .const import API_EMAIL, API_PASSWORD, API_TOKEN, HASS_UUID, SOURCE_USER
from tests.common import MockConfigEntry
async def test_full_flow(
hass: HomeAssistant,
mock_customer_api_client: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test we can finish a config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
hass.data["core.uuid"] = HASS_UUID
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_EMAIL: API_EMAIL,
CONF_PASSWORD: API_PASSWORD,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == API_EMAIL
assert result["data"] == {
CONF_TOKEN: API_TOKEN,
CONF_EMAIL: API_EMAIL,
}
assert len(mock_customer_api_client.mock_calls) == 1
async def test_form_unique_id(
hass: HomeAssistant, mock_customer_api_client: AsyncMock
) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
hass.data["core.uuid"] = HASS_UUID
entry = MockConfigEntry(unique_id="39", domain=DOMAIN)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_EMAIL: API_EMAIL,
CONF_PASSWORD: API_PASSWORD,
},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert len(mock_customer_api_client.mock_calls) == 1
@pytest.mark.parametrize(
("exception", "error"),
[
(StTimeout, "cannot_connect"),
(StInvalidAuth("Invalid credentials"), "invalid_auth"),
(DecodeError("Bad API token"), "invalid_access_token"),
],
)
async def test_form_exceptions(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mock_customer_api_client: AsyncMock,
exception: Exception,
error: str,
) -> None:
"""Test we handle config form exceptions."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
hass.data["core.uuid"] = HASS_UUID
mock_customer_api_client.get_token.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_EMAIL: API_EMAIL,
CONF_PASSWORD: API_PASSWORD,
},
)
assert result["errors"] == {"base": error}
assert result["type"] is FlowResultType.FORM
mock_customer_api_client.get_token.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_EMAIL: API_EMAIL,
CONF_PASSWORD: API_PASSWORD,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == API_EMAIL
assert result["data"] == {
CONF_TOKEN: API_TOKEN,
CONF_EMAIL: API_EMAIL,
}
assert len(mock_customer_api_client.mock_calls) == 2