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

Add a config flow for Recollect Waste (#43063)

Browse Source 
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
Aaron Bach 2020-11-12 03:00:42 -07:00 committed by GitHub
parent cdc53329d0
commit 24840cce23
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 382 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -711,6 +711,7 @@ omit =
homeassistant/components/rainforest_eagle/sensor.py
homeassistant/components/raspihats/*
homeassistant/components/raspyrfm/*
homeassistant/components/recollect_waste/__init__.py
homeassistant/components/recollect_waste/sensor.py
homeassistant/components/recswitch/switch.py
homeassistant/components/reddit/*

1
CODEOWNERS
View File

@ -361,6 +361,7 @@ homeassistant/components/raincloud/* @vanstinator
homeassistant/components/rainforest_eagle/* @gtdiehl @jcalbert
homeassistant/components/rainmachine/* @bachya
homeassistant/components/random/* @fabaff
homeassistant/components/recollect_waste/* @bachya
homeassistant/components/rejseplanen/* @DarkFox
homeassistant/components/repetier/* @MTrab
homeassistant/components/rfxtrx/* @danielhiversen @elupus @RobBie1221

84
homeassistant/components/recollect_waste/__init__.py
View File

@ -1 +1,83 @@
"""The recollect_waste component."""
"""The Recollect Waste integration."""
import asyncio
from datetime import date, timedelta
from typing import List
from aiorecollect.client import Client, PickupEvent
from aiorecollect.errors import RecollectError
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_PLACE_ID, CONF_SERVICE_ID, DATA_COORDINATOR, DOMAIN, LOGGER
DEFAULT_NAME = "recollect_waste"
DEFAULT_UPDATE_INTERVAL = timedelta(days=1)
PLATFORMS = ["sensor"]
async def async_setup(hass: HomeAssistant, config: dict) -> bool:
"""Set up the RainMachine component."""
hass.data[DOMAIN] = {DATA_COORDINATOR: {}}
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up RainMachine as config entry."""
session = aiohttp_client.async_get_clientsession(hass)
client = Client(
entry.data[CONF_PLACE_ID], entry.data[CONF_SERVICE_ID], session=session
)
async def async_get_pickup_events() -> List[PickupEvent]:
"""Get the next pickup."""
try:
return await client.async_get_pickup_events(
start_date=date.today(), end_date=date.today() + timedelta(weeks=4)
)
except RecollectError as err:
raise UpdateFailed(
f"Error while requesting data from Recollect: {err}"
) from err
coordinator = DataUpdateCoordinator(
hass,
LOGGER,
name=f"Place {entry.data[CONF_PLACE_ID]}, Service {entry.data[CONF_SERVICE_ID]}",
update_interval=DEFAULT_UPDATE_INTERVAL,
update_method=async_get_pickup_events,
)
await coordinator.async_refresh()
if not coordinator.last_update_success:
raise ConfigEntryNotReady
hass.data[DOMAIN][DATA_COORDINATOR][entry.entry_id] = coordinator
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload an RainMachine config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)
if unload_ok:
hass.data[DOMAIN][DATA_COORDINATOR].pop(entry.entry_id)
return unload_ok

64
homeassistant/components/recollect_waste/config_flow.py Normal file
View File

@ -0,0 +1,64 @@
"""Config flow for Recollect Waste integration."""
from aiorecollect.client import Client
from aiorecollect.errors import RecollectError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.helpers import aiohttp_client
from .const import ( # pylint:disable=unused-import
CONF_PLACE_ID,
CONF_SERVICE_ID,
DOMAIN,
LOGGER,
)
DATA_SCHEMA = vol.Schema(
{vol.Required(CONF_PLACE_ID): str, vol.Required(CONF_SERVICE_ID): str}
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Recollect Waste."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
async def async_step_import(self, import_config: dict = None) -> dict:
"""Handle configuration via YAML import."""
return await self.async_step_user(import_config)
async def async_step_user(self, user_input: dict = None) -> dict:
"""Handle configuration via the UI."""
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA, errors={}
)
unique_id = f"{user_input[CONF_PLACE_ID]}, {user_input[CONF_SERVICE_ID]}"
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured()
session = aiohttp_client.async_get_clientsession(self.hass)
client = Client(
user_input[CONF_PLACE_ID], user_input[CONF_SERVICE_ID], session=session
)
try:
await client.async_get_next_pickup_event()
except RecollectError as err:
LOGGER.error("Error during setup of integration: %s", err)
return self.async_show_form(
step_id="user",
data_schema=DATA_SCHEMA,
errors={"base": "invalid_place_or_service_id"},
)
return self.async_create_entry(
title=unique_id,
data={
CONF_PLACE_ID: user_input[CONF_PLACE_ID],
CONF_SERVICE_ID: user_input[CONF_SERVICE_ID],
},
)

11
homeassistant/components/recollect_waste/const.py Normal file
View File

@ -0,0 +1,11 @@
"""Define Recollect Waste constants."""
import logging
DOMAIN = "recollect_waste"
LOGGER = logging.getLogger(__package__)
CONF_PLACE_ID = "place_id"
CONF_SERVICE_ID = "service_id"
DATA_COORDINATOR = "coordinator"

9
homeassistant/components/recollect_waste/manifest.json
View File

@ -1,7 +1,12 @@
{
"domain": "recollect_waste",
"name": "ReCollect Waste",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/recollect_waste",
"requirements": ["aiorecollect==0.2.1"],
"codeowners": []
"requirements": [
"aiorecollect==0.2.1"
],
"codeowners": [
"@bachya"
]
}

145
homeassistant/components/recollect_waste/sensor.py
View File

@ -1,27 +1,30 @@
"""Support for Recollect Waste curbside collection pickup."""
from datetime import date, timedelta
import logging
"""Support for Recollect Waste sensors."""
from typing import Callable
from aiorecollect import Client
from aiorecollect.errors import RecollectError
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.const import CONF_NAME
from homeassistant.helpers import aiohttp_client, config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import CONF_PLACE_ID, CONF_SERVICE_ID, DATA_COORDINATOR, DOMAIN, LOGGER
_LOGGER = logging.getLogger(__name__)
ATTR_PICKUP_TYPES = "pickup_types"
ATTR_AREA_NAME = "area_name"
ATTR_NEXT_PICKUP_TYPES = "next_pickup_types"
ATTR_NEXT_PICKUP_DATE = "next_pickup_date"
CONF_PLACE_ID = "place_id"
CONF_SERVICE_ID = "service_id"
DEFAULT_NAME = "recollect_waste"
ICON = "mdi:trash-can-outline"
SCAN_INTERVAL = timedelta(days=1)
DEFAULT_ATTRIBUTION = "Pickup data provided by Recollect Waste"
DEFAULT_NAME = "recollect_waste"
DEFAULT_ICON = "mdi:trash-can-outline"
CONF_NAME = "name"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
@ -32,70 +35,88 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the Recollect Waste platform."""
session = aiohttp_client.async_get_clientsession(hass)
client = Client(config[CONF_PLACE_ID], config[CONF_SERVICE_ID], session=session)
# Ensure the client can connect to the API successfully
# with given place_id and service_id.
try:
await client.async_get_next_pickup_event()
except RecollectError as err:
_LOGGER.error("Error setting up Recollect sensor platform: %s", err)
return
async_add_entities([RecollectWasteSensor(config.get(CONF_NAME), client)], True)
async def async_setup_platform(
hass: HomeAssistant,
config: dict,
async_add_entities: Callable,
discovery_info: dict = None,
):
"""Import Awair configuration from YAML."""
LOGGER.warning(
"Loading Recollect Waste via platform setup is deprecated. "
"Please remove it from your configuration."
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
)
class RecollectWasteSensor(Entity):
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: Callable
) -> None:
"""Set up Recollect Waste sensors based on a config entry."""
coordinator = hass.data[DOMAIN][DATA_COORDINATOR][entry.entry_id]
async_add_entities([RecollectWasteSensor(coordinator, entry)])
class RecollectWasteSensor(CoordinatorEntity):
"""Recollect Waste Sensor."""
def __init__(self, name, client):
def __init__(self, coordinator: DataUpdateCoordinator, entry: ConfigEntry) -> None:
"""Initialize the sensor."""
self._attributes = {}
self._name = name
super().__init__(coordinator)
self._attributes = {ATTR_ATTRIBUTION: DEFAULT_ATTRIBUTION}
self._place_id = entry.data[CONF_PLACE_ID]
self._service_id = entry.data[CONF_SERVICE_ID]
self._state = None
self.client = client
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def unique_id(self) -> str:
"""Return a unique ID."""
return f"{self.client.place_id}{self.client.service_id}"
@property
def state(self):
"""Return the state of the sensor."""
return self._state
@property
def device_state_attributes(self):
def device_state_attributes(self) -> dict:
"""Return the state attributes."""
return self._attributes
@property
def icon(self):
def icon(self) -> str:
"""Icon to use in the frontend."""
return ICON
return DEFAULT_ICON
async def async_update(self):
"""Update device state."""
try:
pickup_event_array = await self.client.async_get_pickup_events(
start_date=date.today(), end_date=date.today() + timedelta(weeks=4)
)
except RecollectError as err:
_LOGGER.error("Error while requesting data from Recollect: %s", err)
return
@property
def name(self) -> str:
"""Return the name of the sensor."""
return DEFAULT_NAME
pickup_event = pickup_event_array[0]
next_pickup_event = pickup_event_array[1]
@property
def state(self) -> str:
"""Return the state of the sensor."""
return self._state
@property
def unique_id(self) -> str:
"""Return a unique ID."""
return f"{self._place_id}{self._service_id}"
@callback
def _handle_coordinator_update(self) -> None:
"""Respond to a DataUpdateCoordinator update."""
self.update_from_latest_data()
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
self.update_from_latest_data()
@callback
def update_from_latest_data(self) -> None:
"""Update the state."""
pickup_event = self.coordinator.data[0]
next_pickup_event = self.coordinator.data[1]
next_date = str(next_pickup_event.date)
self._state = pickup_event.date
self._attributes.update(
{

18
homeassistant/components/recollect_waste/strings.json Normal file
View File

@ -0,0 +1,18 @@
{
"config": {
"step": {
"user": {
"data": {
"place_id": "Place ID",
"service_id": "Service ID"
}
}
},
"error": {
"invalid_place_or_service_id": "Invalid Place or Service ID"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
}
}

18
homeassistant/components/recollect_waste/translations/en.json Normal file
View File

@ -0,0 +1,18 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured"
},
"error": {
"invalid_place_or_service_id": "Invalid Place or Service ID"
},
"step": {
"user": {
"data": {
"place_id": "Place ID",
"service_id": "Service ID"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -157,6 +157,7 @@ FLOWS = [
"pvpc_hourly_pricing",
"rachio",
"rainmachine",
"recollect_waste",
"rfxtrx",
"ring",
"risco",

3
requirements_test_all.txt
View File

@ -136,6 +136,9 @@ aiopvpc==2.0.2
# homeassistant.components.webostv
aiopylgtv==0.3.3
# homeassistant.components.recollect_waste
aiorecollect==0.2.1
# homeassistant.components.shelly
aioshelly==0.5.1

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

@ -0,0 +1 @@
"""Define tests for the Recollet Waste integration."""

91
tests/components/recollect_waste/test_config_flow.py Normal file
View File

@ -0,0 +1,91 @@
"""Define tests for the Recollect Waste config flow."""
from aiorecollect.errors import RecollectError
from homeassistant import data_entry_flow
from homeassistant.components.recollect_waste import (
CONF_PLACE_ID,
CONF_SERVICE_ID,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from tests.async_mock import patch
from tests.common import MockConfigEntry
async def test_duplicate_error(hass):
"""Test that errors are shown when duplicates are added."""
conf = {CONF_PLACE_ID: "12345", CONF_SERVICE_ID: "12345"}
MockConfigEntry(domain=DOMAIN, unique_id="12345, 12345", data=conf).add_to_hass(
hass
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_invalid_place_or_service_id(hass):
"""Test that an invalid Place or Service ID throws an error."""
conf = {CONF_PLACE_ID: "12345", CONF_SERVICE_ID: "12345"}
with patch(
"aiorecollect.client.Client.async_get_next_pickup_event",
side_effect=RecollectError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "invalid_place_or_service_id"}
async def test_show_form(hass):
"""Test that the form is served with no input."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
async def test_step_import(hass):
"""Test that the user step works."""
conf = {CONF_PLACE_ID: "12345", CONF_SERVICE_ID: "12345"}
with patch(
"homeassistant.components.recollect_waste.async_setup_entry", return_value=True
), patch(
"aiorecollect.client.Client.async_get_next_pickup_event", return_value=True
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=conf
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "12345, 12345"
assert result["data"] == {CONF_PLACE_ID: "12345", CONF_SERVICE_ID: "12345"}
async def test_step_user(hass):
"""Test that the user step works."""
conf = {CONF_PLACE_ID: "12345", CONF_SERVICE_ID: "12345"}
with patch(
"homeassistant.components.recollect_waste.async_setup_entry", return_value=True
), patch(
"aiorecollect.client.Client.async_get_next_pickup_event", return_value=True
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=conf
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "12345, 12345"
assert result["data"] == {CONF_PLACE_ID: "12345", CONF_SERVICE_ID: "12345"}