Flipr integration (#46582)

Co-authored-by: Franck Nijhof <frenck@frenck.nl>
Co-authored-by: cnico <>
This commit is contained in:
cnico 2021-07-21 21:35:44 +02:00 committed by GitHub
parent 3eb3c2824c
commit 6636e5b737
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 681 additions and 0 deletions

View File

@ -161,6 +161,7 @@ homeassistant/components/fireservicerota/* @cyberjunky
homeassistant/components/firmata/* @DaAwesomeP
homeassistant/components/fixer/* @fabaff
homeassistant/components/flick_electric/* @ZephireNZ
homeassistant/components/flipr/* @cnico
homeassistant/components/flo/* @dmulcahey
homeassistant/components/flock/* @fabaff
homeassistant/components/flume/* @ChrisMandich @bdraco

View File

@ -0,0 +1,90 @@
"""The Flipr integration."""
from datetime import timedelta
import logging
from flipr_api import FliprAPIRestClient
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import CONF_FLIPR_ID, DOMAIN, MANUFACTURER, NAME
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(minutes=60)
PLATFORMS = ["sensor"]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Flipr from a config entry."""
hass.data.setdefault(DOMAIN, {})
coordinator = FliprDataUpdateCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN][entry.entry_id] = coordinator
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
class FliprDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to hold Flipr data retrieval."""
def __init__(self, hass, entry):
"""Initialize."""
username = entry.data[CONF_EMAIL]
password = entry.data[CONF_PASSWORD]
self.flipr_id = entry.data[CONF_FLIPR_ID]
_LOGGER.debug("Config entry values : %s, %s", username, self.flipr_id)
# Establishes the connection.
self.client = FliprAPIRestClient(username, password)
self.entry = entry
super().__init__(
hass,
_LOGGER,
name=f"Flipr data measure for {self.flipr_id}",
update_interval=SCAN_INTERVAL,
)
async def _async_update_data(self):
"""Fetch data from API endpoint."""
return await self.hass.async_add_executor_job(
self.client.get_pool_measure_latest, self.flipr_id
)
class FliprEntity(CoordinatorEntity):
"""Implements a common class elements representing the Flipr component."""
def __init__(self, coordinator, flipr_id, info_type):
"""Initialize Flipr sensor."""
super().__init__(coordinator)
self._attr_unique_id = f"{flipr_id}-{info_type}"
self._attr_device_info = {
"identifiers": {(DOMAIN, flipr_id)},
"name": NAME,
"manufacturer": MANUFACTURER,
}
self.info_type = info_type
self.flipr_id = flipr_id

View File

@ -0,0 +1,124 @@
"""Config flow for Flipr integration."""
from __future__ import annotations
import logging
from flipr_api import FliprAPIRestClient
from requests.exceptions import HTTPError, Timeout
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from .const import CONF_FLIPR_ID, DOMAIN
_LOGGER = logging.getLogger(__name__)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Flipr."""
VERSION = 1
_username: str | None = None
_password: str | None = None
_flipr_id: str | None = None
_possible_flipr_ids: list[str] | None = None
async def async_step_user(self, user_input=None):
"""Handle the initial step."""
if user_input is None:
return self._show_setup_form()
self._username = user_input[CONF_EMAIL]
self._password = user_input[CONF_PASSWORD]
errors = {}
if not self._flipr_id:
try:
flipr_ids = await self._authenticate_and_search_flipr()
except HTTPError:
errors["base"] = "invalid_auth"
except (Timeout, ConnectionError):
errors["base"] = "cannot_connect"
except Exception as exception: # pylint: disable=broad-except
errors["base"] = "unknown"
_LOGGER.exception(exception)
if not errors and len(flipr_ids) == 0:
# No flipr_id found. Tell the user with an error message.
errors["base"] = "no_flipr_id_found"
if errors:
return self._show_setup_form(errors)
if len(flipr_ids) == 1:
self._flipr_id = flipr_ids[0]
else:
# If multiple flipr found (rare case), we ask the user to choose one in a select box.
# The user will have to run config_flow as many times as many fliprs he has.
self._possible_flipr_ids = flipr_ids
return await self.async_step_flipr_id()
# Check if already configured
await self.async_set_unique_id(self._flipr_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=self._flipr_id,
data={
CONF_EMAIL: self._username,
CONF_PASSWORD: self._password,
CONF_FLIPR_ID: self._flipr_id,
},
)
def _show_setup_form(self, errors=None):
"""Show the setup form to the user."""
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_EMAIL): str, vol.Required(CONF_PASSWORD): str}
),
errors=errors,
)
async def _authenticate_and_search_flipr(self) -> list[str]:
"""Validate the username and password provided and searches for a flipr id."""
client = await self.hass.async_add_executor_job(
FliprAPIRestClient, self._username, self._password
)
flipr_ids = await self.hass.async_add_executor_job(client.search_flipr_ids)
return flipr_ids
async def async_step_flipr_id(self, user_input=None):
"""Handle the initial step."""
if not user_input:
# Creation of a select with the proposal of flipr ids values found by API.
flipr_ids_for_form = {}
for flipr_id in self._possible_flipr_ids:
flipr_ids_for_form[flipr_id] = f"{flipr_id}"
return self.async_show_form(
step_id="flipr_id",
data_schema=vol.Schema(
{
vol.Required(CONF_FLIPR_ID): vol.All(
vol.Coerce(str), vol.In(flipr_ids_for_form)
)
}
),
)
# Get chosen flipr_id.
self._flipr_id = user_input[CONF_FLIPR_ID]
return await self.async_step_user(
{
CONF_EMAIL: self._username,
CONF_PASSWORD: self._password,
CONF_FLIPR_ID: self._flipr_id,
}
)

View File

@ -0,0 +1,10 @@
"""Constants for the Flipr integration."""
DOMAIN = "flipr"
CONF_FLIPR_ID = "flipr_id"
ATTRIBUTION = "Flipr Data"
MANUFACTURER = "CTAC-TECH"
NAME = "Flipr"

View File

@ -0,0 +1,12 @@
{
"domain": "flipr",
"name": "Flipr",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/flipr",
"requirements": [
"flipr-api==1.4.1"],
"codeowners": [
"@cnico"
],
"iot_class": "cloud_polling"
}

View File

@ -0,0 +1,90 @@
"""Sensor platform for the Flipr's pool_sensor."""
from datetime import datetime
from homeassistant.const import (
ATTR_ATTRIBUTION,
DEVICE_CLASS_TEMPERATURE,
DEVICE_CLASS_TIMESTAMP,
TEMP_CELSIUS,
)
from homeassistant.helpers.entity import Entity
from . import FliprEntity
from .const import ATTRIBUTION, CONF_FLIPR_ID, DOMAIN
SENSORS = {
"chlorine": {
"unit": "mV",
"icon": "mdi:pool",
"name": "Chlorine",
"device_class": None,
},
"ph": {"unit": None, "icon": "mdi:pool", "name": "pH", "device_class": None},
"temperature": {
"unit": TEMP_CELSIUS,
"icon": None,
"name": "Water Temp",
"device_class": DEVICE_CLASS_TEMPERATURE,
},
"date_time": {
"unit": None,
"icon": None,
"name": "Last Measured",
"device_class": DEVICE_CLASS_TIMESTAMP,
},
"red_ox": {
"unit": "mV",
"icon": "mdi:pool",
"name": "Red OX",
"device_class": None,
},
}
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Defer sensor setup to the shared sensor module."""
flipr_id = config_entry.data[CONF_FLIPR_ID]
coordinator = hass.data[DOMAIN][config_entry.entry_id]
sensors_list = []
for sensor in SENSORS:
sensors_list.append(FliprSensor(coordinator, flipr_id, sensor))
async_add_entities(sensors_list, True)
class FliprSensor(FliprEntity, Entity):
"""Sensor representing FliprSensor data."""
@property
def name(self):
"""Return the name of the particular component."""
return f"Flipr {self.flipr_id} {SENSORS[self.info_type]['name']}"
@property
def state(self):
"""State of the sensor."""
state = self.coordinator.data[self.info_type]
if isinstance(state, datetime):
return state.isoformat()
return state
@property
def device_class(self):
"""Return the device class."""
return SENSORS[self.info_type]["device_class"]
@property
def icon(self):
"""Return the icon."""
return SENSORS[self.info_type]["icon"]
@property
def unit_of_measurement(self):
"""Return unit of measurement."""
return SENSORS[self.info_type]["unit"]
@property
def device_state_attributes(self):
"""Return device attributes."""
return {ATTR_ATTRIBUTION: ATTRIBUTION}

View File

@ -0,0 +1,30 @@
{
"config": {
"step": {
"user": {
"title": "Connect to Flipr",
"description": "Connect using your Flipr account.",
"data": {
"email": "[%key:common::config_flow::data::email%]",
"password": "[%key:common::config_flow::data::password%]"
}
},
"flipr_id": {
"title": "Choose your Flipr",
"description": "Choose your Flipr ID in the list",
"data": {
"flipr_id": "Flipr ID"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"no_flipr_id_found": "No flipr id associated to your account for now. You should verify it is working with the Flipr's mobile app first."
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

View File

@ -0,0 +1,30 @@
{
"config": {
"abort": {
"already_configured": "This Flipr is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"unknown": "Unexpected error",
"no_flipr_id_found": "No flipr id associated to your account for now. You should verify it is working with the Flipr's mobile app first."
},
"step": {
"user": {
"data": {
"email": "Email",
"password": "Password"
},
"description": "Connect to your flipr account",
"title": "Flipr device"
},
"flipr_id": {
"data": {
"flipr_id": "Flipr ID"
},
"description": "Choose your flipr ID in the list",
"title": "Flipr device"
}
}
}
}

View File

@ -77,6 +77,7 @@ FLOWS = [
"faa_delays",
"fireservicerota",
"flick_electric",
"flipr",
"flo",
"flume",
"flunearyou",

View File

@ -617,6 +617,9 @@ fitbit==0.3.1
# homeassistant.components.fixer
fixerio==1.0.0a0
# homeassistant.components.flipr
flipr-api==1.4.1
# homeassistant.components.flux_led
flux_led==0.22

View File

@ -338,6 +338,9 @@ faadelays==0.0.7
# homeassistant.components.feedreader
feedparser==6.0.2
# homeassistant.components.flipr
flipr-api==1.4.1
# homeassistant.components.homekit
fnvhash==0.1.0

View File

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

View File

@ -0,0 +1,166 @@
"""Test the Flipr config flow."""
from unittest.mock import patch
import pytest
from requests.exceptions import HTTPError, Timeout
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.components.flipr.const import CONF_FLIPR_ID, DOMAIN
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
@pytest.fixture(name="mock_setup")
def mock_setups():
"""Prevent setup."""
with patch(
"homeassistant.components.flipr.async_setup_entry",
return_value=True,
):
yield
async def test_show_form(hass):
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
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["step_id"] == config_entries.SOURCE_USER
async def test_invalid_credential(hass, mock_setup):
"""Test invalid credential."""
with patch(
"flipr_api.FliprAPIRestClient.search_flipr_ids", side_effect=HTTPError()
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={
CONF_EMAIL: "bad_login",
CONF_PASSWORD: "bad_pass",
CONF_FLIPR_ID: "",
},
)
assert result["type"] == "form"
assert result["errors"] == {"base": "invalid_auth"}
async def test_nominal_case(hass, mock_setup):
"""Test valid login form."""
with patch(
"flipr_api.FliprAPIRestClient.search_flipr_ids",
return_value=["flipid"],
) as mock_flipr_client:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
CONF_FLIPR_ID: "flipid",
},
)
await hass.async_block_till_done()
assert len(mock_flipr_client.mock_calls) == 1
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "flipid"
assert result["data"] == {
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
CONF_FLIPR_ID: "flipid",
}
async def test_multiple_flip_id(hass, mock_setup):
"""Test multiple flipr id adding a config step."""
with patch(
"flipr_api.FliprAPIRestClient.search_flipr_ids",
return_value=["FLIP1", "FLIP2"],
) as mock_flipr_client:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "flipr_id"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_FLIPR_ID: "FLIP2"},
)
assert len(mock_flipr_client.mock_calls) == 1
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "FLIP2"
assert result["data"] == {
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
CONF_FLIPR_ID: "FLIP2",
}
async def test_no_flip_id(hass, mock_setup):
"""Test no flipr id found."""
with patch(
"flipr_api.FliprAPIRestClient.search_flipr_ids",
return_value=[],
) as mock_flipr_client:
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
},
)
assert result["step_id"] == "user"
assert result["type"] == "form"
assert result["errors"] == {"base": "no_flipr_id_found"}
assert len(mock_flipr_client.mock_calls) == 1
async def test_http_errors(hass, mock_setup):
"""Test HTTP Errors."""
with patch("flipr_api.FliprAPIRestClient.search_flipr_ids", side_effect=Timeout()):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={
CONF_EMAIL: "nada",
CONF_PASSWORD: "nada",
CONF_FLIPR_ID: "",
},
)
assert result["type"] == "form"
assert result["errors"] == {"base": "cannot_connect"}
with patch(
"flipr_api.FliprAPIRestClient.search_flipr_ids",
side_effect=Exception("Bad request Boy :) --"),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={
CONF_EMAIL: "nada",
CONF_PASSWORD: "nada",
CONF_FLIPR_ID: "",
},
)
assert result["type"] == "form"
assert result["errors"] == {"base": "unknown"}

View File

@ -0,0 +1,28 @@
"""Tests for init methods."""
from unittest.mock import patch
from homeassistant.components.flipr.const import CONF_FLIPR_ID, DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_unload_entry(hass: HomeAssistant):
"""Test unload entry."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_EMAIL: "dummylogin",
CONF_PASSWORD: "dummypass",
CONF_FLIPR_ID: "FLIP1",
},
unique_id="123456",
)
entry.add_to_hass(hass)
with patch("homeassistant.components.flipr.FliprAPIRestClient"):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await hass.config_entries.async_unload(entry.entry_id)
assert entry.state == ConfigEntryState.NOT_LOADED

View File

@ -0,0 +1,92 @@
"""Test the Flipr sensor and binary sensor."""
from datetime import datetime
from unittest.mock import patch
from homeassistant.components.flipr.const import CONF_FLIPR_ID, DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import (
ATTR_ICON,
ATTR_UNIT_OF_MEASUREMENT,
CONF_EMAIL,
CONF_PASSWORD,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from homeassistant.util import dt as dt_util
from tests.common import MockConfigEntry
# Data for the mocked object returned via flipr_api client.
MOCK_DATE_TIME = datetime(2021, 2, 15, 9, 10, 32, tzinfo=dt_util.UTC)
MOCK_FLIPR_MEASURE = {
"temperature": 10.5,
"ph": 7.03,
"chlorine": 0.23654886,
"red_ox": 657.58,
"date_time": MOCK_DATE_TIME,
"ph_status": "TooLow",
"chlorine_status": "Medium",
}
async def test_sensors(hass: HomeAssistant) -> None:
"""Test the creation and values of the Flipr sensors."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="test_entry_unique_id",
data={
CONF_EMAIL: "toto@toto.com",
CONF_PASSWORD: "myPassword",
CONF_FLIPR_ID: "myfliprid",
},
)
entry.add_to_hass(hass)
registry = await hass.helpers.entity_registry.async_get_registry()
# Pre-create registry entries for sensors
registry.async_get_or_create(
SENSOR_DOMAIN,
DOMAIN,
"my_random_entity_id",
suggested_object_id="sensor.flipr_myfliprid_chlorine",
disabled_by=None,
)
with patch(
"flipr_api.FliprAPIRestClient.get_pool_measure_latest",
return_value=MOCK_FLIPR_MEASURE,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get("sensor.flipr_myfliprid_ph")
assert state
assert state.attributes.get(ATTR_ICON) == "mdi:pool"
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
assert state.state == "7.03"
state = hass.states.get("sensor.flipr_myfliprid_water_temp")
assert state
assert state.attributes.get(ATTR_ICON) is None
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is TEMP_CELSIUS
assert state.state == "10.5"
state = hass.states.get("sensor.flipr_myfliprid_last_measured")
assert state
assert state.attributes.get(ATTR_ICON) is None
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
assert state.state == "2021-02-15T09:10:32+00:00"
state = hass.states.get("sensor.flipr_myfliprid_red_ox")
assert state
assert state.attributes.get(ATTR_ICON) == "mdi:pool"
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == "mV"
assert state.state == "657.58"
state = hass.states.get("sensor.flipr_myfliprid_chlorine")
assert state
assert state.attributes.get(ATTR_ICON) == "mdi:pool"
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == "mV"
assert state.state == "0.23654886"