Add Garages Amsterdam integration (#43157)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
Co-authored-by: Franck Nijhof <git@frenck.dev>
Co-authored-by: Franck Nijhof <frenck@frenck.nl>
This commit is contained in:
Klaas Schoute 2021-05-15 20:43:12 +02:00 committed by GitHub
parent 0c37effc72
commit e1dd479e15
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 451 additions and 0 deletions

View file

@ -348,6 +348,9 @@ omit =
homeassistant/components/frontier_silicon/media_player.py
homeassistant/components/futurenow/light.py
homeassistant/components/garadget/cover.py
homeassistant/components/garages_amsterdam/__init__.py
homeassistant/components/garages_amsterdam/binary_sensor.py
homeassistant/components/garages_amsterdam/sensor.py
homeassistant/components/garmin_connect/__init__.py
homeassistant/components/garmin_connect/const.py
homeassistant/components/garmin_connect/sensor.py

View file

@ -170,6 +170,7 @@ homeassistant/components/fritz/* @mammuth @AaronDavidSchneider @chemelli74
homeassistant/components/fritzbox/* @mib1185
homeassistant/components/fronius/* @nielstron
homeassistant/components/frontend/* @home-assistant/frontend
homeassistant/components/garages_amsterdam/* @klaasnicolaas
homeassistant/components/garmin_connect/* @cyberjunky
homeassistant/components/gdacs/* @exxamalte
homeassistant/components/geniushub/* @zxdavb

View file

@ -0,0 +1,60 @@
"""The Garages Amsterdam integration."""
from datetime import timedelta
import logging
import async_timeout
import garages_amsterdam
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
PLATFORMS = ["binary_sensor", "sensor"]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Garages Amsterdam from a config entry."""
await get_coordinator(hass)
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload Garages Amsterdam config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if len(hass.config_entries.async_entries(DOMAIN)) == 1:
hass.data.pop(DOMAIN)
return unload_ok
async def get_coordinator(
hass: HomeAssistant,
) -> DataUpdateCoordinator:
"""Get the data update coordinator."""
if DOMAIN in hass.data:
return hass.data[DOMAIN]
async def async_get_garages():
with async_timeout.timeout(10):
return {
garage.garage_name: garage
for garage in await garages_amsterdam.get_garages(
aiohttp_client.async_get_clientsession(hass)
)
}
coordinator = DataUpdateCoordinator(
hass,
logging.getLogger(__name__),
name=DOMAIN,
update_method=async_get_garages,
update_interval=timedelta(minutes=10),
)
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN] = coordinator
return coordinator

View file

@ -0,0 +1,81 @@
"""Binary Sensor platform for Garages Amsterdam."""
from __future__ import annotations
from typing import Any
from homeassistant.components.binary_sensor import (
DEVICE_CLASS_PROBLEM,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from . import get_coordinator
from .const import ATTRIBUTION
BINARY_SENSORS = {
"state",
}
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Defer sensor setup to the shared sensor module."""
coordinator = await get_coordinator(hass)
async_add_entities(
GaragesamsterdamBinarySensor(
coordinator, config_entry.data["garage_name"], info_type
)
for info_type in BINARY_SENSORS
)
class GaragesamsterdamBinarySensor(CoordinatorEntity, BinarySensorEntity):
"""Binary Sensor representing garages amsterdam data."""
def __init__(
self, coordinator: DataUpdateCoordinator, garage_name: str, info_type: str
) -> None:
"""Initialize garages amsterdam binary sensor."""
super().__init__(coordinator)
self._unique_id = f"{garage_name}-{info_type}"
self._garage_name = garage_name
self._info_type = info_type
self._name = garage_name
@property
def name(self) -> str:
"""Return the name of the sensor."""
return self._name
@property
def unique_id(self) -> str:
"""Return the unique id of the device."""
return self._unique_id
@property
def is_on(self) -> bool:
"""If the binary sensor is currently on or off."""
return (
getattr(self.coordinator.data[self._garage_name], self._info_type) != "ok"
)
@property
def device_class(self) -> str:
"""Return the class of the binary sensor."""
return DEVICE_CLASS_PROBLEM
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return device attributes."""
return {ATTR_ATTRIBUTION: ATTRIBUTION}

View file

@ -0,0 +1,58 @@
"""Config flow for Garages Amsterdam integration."""
from __future__ import annotations
import logging
from typing import Any
from aiohttp import ClientResponseError
import garages_amsterdam
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Garages Amsterdam."""
VERSION = 1
_options: list[str] | None = None
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if self._options is None:
self._options = []
try:
api_data = await garages_amsterdam.get_garages(
aiohttp_client.async_get_clientsession(self.hass)
)
except ClientResponseError:
_LOGGER.error("Unexpected response from server")
return self.async_abort(reason="cannot_connect")
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
for garage in sorted(api_data, key=lambda garage: garage.garage_name):
self._options.append(garage.garage_name)
if user_input is not None:
await self.async_set_unique_id(user_input["garage_name"])
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input["garage_name"], data=user_input
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required("garage_name"): vol.In(self._options)}
),
)

View file

@ -0,0 +1,4 @@
"""Constants for the Garages Amsterdam integration."""
DOMAIN = "garages_amsterdam"
ATTRIBUTION = f'{"Data provided by municipality of Amsterdam"}'

View file

@ -0,0 +1,9 @@
{
"domain": "garages_amsterdam",
"name": "Garages Amsterdam",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/garages_amsterdam",
"requirements": ["garages-amsterdam==2.0.4"],
"codeowners": ["@klaasnicolaas"],
"iot_class": "cloud_polling"
}

View file

@ -0,0 +1,96 @@
"""Sensor platform for Garages Amsterdam."""
from __future__ import annotations
from typing import Any
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ATTRIBUTION
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from . import get_coordinator
from .const import ATTRIBUTION
SENSORS = {
"free_space_short": "mdi:car",
"free_space_long": "mdi:car",
"short_capacity": "mdi:car",
"long_capacity": "mdi:car",
}
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Defer sensor setup to the shared sensor module."""
coordinator = await get_coordinator(hass)
entities: list[GaragesamsterdamSensor] = []
for info_type in SENSORS:
if getattr(coordinator.data[config_entry.data["garage_name"]], info_type) != "":
entities.append(
GaragesamsterdamSensor(
coordinator, config_entry.data["garage_name"], info_type
)
)
async_add_entities(entities)
class GaragesamsterdamSensor(CoordinatorEntity, SensorEntity):
"""Sensor representing garages amsterdam data."""
def __init__(
self, coordinator: DataUpdateCoordinator, garage_name: str, info_type: str
) -> None:
"""Initialize garages amsterdam sensor."""
super().__init__(coordinator)
self._unique_id = f"{garage_name}-{info_type}"
self._garage_name = garage_name
self._info_type = info_type
self._name = f"{garage_name} - {info_type}".replace("_", " ")
@property
def name(self) -> str:
"""Return the name of the sensor."""
return self._name
@property
def unique_id(self) -> str:
"""Return the unique id of the device."""
return self._unique_id
@property
def available(self) -> bool:
"""Return if sensor is available."""
return self.coordinator.last_update_success and (
self._garage_name in self.coordinator.data
)
@property
def state(self) -> str:
"""Return the state of the sensor."""
return getattr(self.coordinator.data[self._garage_name], self._info_type)
@property
def icon(self) -> str:
"""Return the icon."""
return SENSORS[self._info_type]
@property
def unit_of_measurement(self) -> str:
"""Return unit of measurement."""
return "cars"
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return device attributes."""
return {ATTR_ATTRIBUTION: ATTRIBUTION}

View file

@ -0,0 +1,16 @@
{
"title": "Garages Amsterdam",
"config": {
"step": {
"user": {
"title": "Pick a garage to monitor",
"data": { "garage_name": "Garage name" }
}
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
}
}
}

View file

@ -0,0 +1,18 @@
{
"config": {
"abort": {
"already_configured": "Device is already configured",
"cannot_connect": "Failed to connect",
"unknown": "Unexpected error"
},
"step": {
"user": {
"data": {
"garage_name": "Garage name"
},
"title": "Pick a garage to monitor"
}
}
},
"title": "Garages Amsterdam"
}

View file

@ -81,6 +81,7 @@ FLOWS = [
"fritz",
"fritzbox",
"fritzbox_callmonitor",
"garages_amsterdam",
"garmin_connect",
"gdacs",
"geofency",

View file

@ -634,6 +634,9 @@ fritzconnection==1.4.2
# homeassistant.components.google_translate
gTTS==2.2.2
# homeassistant.components.garages_amsterdam
garages-amsterdam==2.0.4
# homeassistant.components.garmin_connect
garminconnect==0.1.19

View file

@ -340,6 +340,9 @@ fritzconnection==1.4.2
# homeassistant.components.google_translate
gTTS==2.2.2
# homeassistant.components.garages_amsterdam
garages-amsterdam==2.0.4
# homeassistant.components.garmin_connect
garminconnect==0.1.19

View file

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

View file

@ -0,0 +1,32 @@
"""Test helpers."""
from unittest.mock import Mock, patch
import pytest
@pytest.fixture(autouse=True)
def mock_cases():
"""Mock garages_amsterdam garages."""
with patch(
"garages_amsterdam.get_garages",
return_value=[
Mock(
garage_name="IJDok",
free_space_short=100,
free_space_long=10,
short_capacity=120,
long_capacity=60,
state="ok",
),
Mock(
garage_name="Arena",
free_space_short=200,
free_space_long=20,
short_capacity=240,
long_capacity=80,
state="error",
),
],
) as mock_get_garages:
yield mock_get_garages

View file

@ -0,0 +1,65 @@
"""Test the Garages Amsterdam config flow."""
from unittest.mock import patch
from aiohttp import ClientResponseError
import pytest
from homeassistant import config_entries, setup
from homeassistant.components.garages_amsterdam.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
async def test_full_flow(hass: HomeAssistant) -> None:
"""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.get("type") == RESULT_TYPE_FORM
assert "flow_id" in result
with patch(
"homeassistant.components.garages_amsterdam.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"garage_name": "IJDok"},
)
await hass.async_block_till_done()
assert result2.get("type") == RESULT_TYPE_CREATE_ENTRY
assert result2.get("title") == "IJDok"
assert "result" in result2
assert result2["result"].unique_id == "IJDok"
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
"side_effect,reason",
[
(RuntimeError, "unknown"),
(ClientResponseError(None, None, status=500), "cannot_connect"),
],
)
async def test_error_handling(
side_effect: Exception, reason: str, hass: HomeAssistant
) -> None:
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
with patch(
"homeassistant.components.garages_amsterdam.config_flow.garages_amsterdam.get_garages",
side_effect=side_effect,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") == RESULT_TYPE_ABORT
assert result.get("reason") == reason