Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add sensors for next Picnic deliveries (#66474)

pull/67006/head
corneyl 2022-02-21 23:45:30 +01:00 committed by GitHub
parent e6af7847fc
commit 137793c067
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 164 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
homeassistant/components/picnic/const.py
View File

@ -22,6 +22,7 @@ ATTRIBUTION = "Data provided by Picnic"
ADDRESS = "address"
CART_DATA = "cart_data"
SLOT_DATA = "slot_data"
NEXT_DELIVERY_DATA = "next_delivery_data"
LAST_ORDER_DATA = "last_order_data"
SENSOR_CART_ITEMS_COUNT = "cart_items_count"
@ -33,18 +34,22 @@ SENSOR_SELECTED_SLOT_MIN_ORDER_VALUE = "selected_slot_min_order_value"
SENSOR_LAST_ORDER_SLOT_START = "last_order_slot_start"
SENSOR_LAST_ORDER_SLOT_END = "last_order_slot_end"
SENSOR_LAST_ORDER_STATUS = "last_order_status"
SENSOR_LAST_ORDER_ETA_START = "last_order_eta_start"
SENSOR_LAST_ORDER_ETA_END = "last_order_eta_end"
SENSOR_LAST_ORDER_MAX_ORDER_TIME = "last_order_max_order_time"
SENSOR_LAST_ORDER_DELIVERY_TIME = "last_order_delivery_time"
SENSOR_LAST_ORDER_TOTAL_PRICE = "last_order_total_price"
SENSOR_NEXT_DELIVERY_ETA_START = "next_delivery_eta_start"
SENSOR_NEXT_DELIVERY_ETA_END = "next_delivery_eta_end"
SENSOR_NEXT_DELIVERY_SLOT_START = "next_delivery_slot_start"
SENSOR_NEXT_DELIVERY_SLOT_END = "next_delivery_slot_end"
@dataclass
class PicnicRequiredKeysMixin:
"""Mixin for required keys."""
data_type: Literal["cart_data", "slot_data", "last_order_data"]
data_type: Literal[
"cart_data", "slot_data", "next_delivery_data", "last_order_data"
]
value_fn: Callable[[Any], StateType | datetime]
@ -130,26 +135,6 @@ SENSOR_TYPES: tuple[PicnicSensorEntityDescription, ...] = (
data_type="last_order_data",
value_fn=lambda last_order: last_order.get("status"),
),
PicnicSensorEntityDescription(
key=SENSOR_LAST_ORDER_ETA_START,
device_class=SensorDeviceClass.TIMESTAMP,
icon="mdi:clock-start",
entity_registry_enabled_default=True,
data_type="last_order_data",
value_fn=lambda last_order: dt_util.parse_datetime(
str(last_order.get("eta", {}).get("start"))
),
),
PicnicSensorEntityDescription(
key=SENSOR_LAST_ORDER_ETA_END,
device_class=SensorDeviceClass.TIMESTAMP,
icon="mdi:clock-end",
entity_registry_enabled_default=True,
data_type="last_order_data",
value_fn=lambda last_order: dt_util.parse_datetime(
str(last_order.get("eta", {}).get("end"))
),
),
PicnicSensorEntityDescription(
key=SENSOR_LAST_ORDER_MAX_ORDER_TIME,
device_class=SensorDeviceClass.TIMESTAMP,
@ -177,4 +162,42 @@ SENSOR_TYPES: tuple[PicnicSensorEntityDescription, ...] = (
data_type="last_order_data",
value_fn=lambda last_order: last_order.get("total_price", 0) / 100,
),
PicnicSensorEntityDescription(
key=SENSOR_NEXT_DELIVERY_ETA_START,
device_class=SensorDeviceClass.TIMESTAMP,
icon="mdi:clock-start",
entity_registry_enabled_default=True,
data_type="next_delivery_data",
value_fn=lambda next_delivery: dt_util.parse_datetime(
str(next_delivery.get("eta", {}).get("start"))
),
),
PicnicSensorEntityDescription(
key=SENSOR_NEXT_DELIVERY_ETA_END,
device_class=SensorDeviceClass.TIMESTAMP,
icon="mdi:clock-end",
entity_registry_enabled_default=True,
data_type="next_delivery_data",
value_fn=lambda next_delivery: dt_util.parse_datetime(
str(next_delivery.get("eta", {}).get("end"))
),
),
PicnicSensorEntityDescription(
key=SENSOR_NEXT_DELIVERY_SLOT_START,
device_class=SensorDeviceClass.TIMESTAMP,
icon="mdi:calendar-start",
data_type="next_delivery_data",
value_fn=lambda next_delivery: dt_util.parse_datetime(
str(next_delivery.get("slot", {}).get("window_start"))
),
),
PicnicSensorEntityDescription(
key=SENSOR_NEXT_DELIVERY_SLOT_END,
device_class=SensorDeviceClass.TIMESTAMP,
icon="mdi:calendar-end",
data_type="next_delivery_data",
value_fn=lambda next_delivery: dt_util.parse_datetime(
str(next_delivery.get("slot", {}).get("window_end"))
),
),
)

45
homeassistant/components/picnic/coordinator.py
View File

@ -13,7 +13,7 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import ADDRESS, CART_DATA, LAST_ORDER_DATA, SLOT_DATA
from .const import ADDRESS, CART_DATA, LAST_ORDER_DATA, NEXT_DELIVERY_DATA, SLOT_DATA
class PicnicUpdateCoordinator(DataUpdateCoordinator):
@ -62,13 +62,14 @@ class PicnicUpdateCoordinator(DataUpdateCoordinator):
if not (cart := self.picnic_api_client.get_cart()):
raise UpdateFailed("API response doesn't contain expected data.")
last_order = self._get_last_order()
next_delivery, last_order = self._get_order_data()
slot_data = self._get_slot_data(cart)
return {
ADDRESS: self._get_address(),
CART_DATA: cart,
SLOT_DATA: slot_data,
NEXT_DELIVERY_DATA: next_delivery,
LAST_ORDER_DATA: last_order,
}
@ -96,47 +97,55 @@ class PicnicUpdateCoordinator(DataUpdateCoordinator):
return {}
def _get_last_order(self) -> dict:
def _get_order_data(self) -> tuple[dict, dict]:
"""Get data of the last order from the list of deliveries."""
# Get the deliveries
deliveries = self.picnic_api_client.get_deliveries(summary=True)
# Determine the last order and return an empty dict if there is none
try:
# Filter on status CURRENT and select the last on the list which is the first one to be delivered
# Make a deepcopy because some references are local
next_deliveries = list(
filter(lambda d: d["status"] == "CURRENT", deliveries)
)
next_delivery = (
copy.deepcopy(next_deliveries[-1]) if next_deliveries else {}
)
last_order = copy.deepcopy(deliveries[0])
except KeyError:
return {}
except (KeyError, TypeError):
# A KeyError or TypeError indicate that the response contains unexpected data
return {}, {}
# Get the position details if the order is not delivered yet
# Get the next order's position details if there is an undelivered order
delivery_position = {}
if not last_order.get("delivery_time"):
if next_delivery and not next_delivery.get("delivery_time"):
try:
delivery_position = self.picnic_api_client.get_delivery_position(
last_order["delivery_id"]
next_delivery["delivery_id"]
)
except ValueError:
# No information yet can mean an empty response
pass
# Determine the ETA, if available, the one from the delivery position API is more precise
# but it's only available shortly before the actual delivery.
last_order["eta"] = delivery_position.get(
"eta_window", last_order.get("eta2", {})
# but, it's only available shortly before the actual delivery.
next_delivery["eta"] = delivery_position.get(
"eta_window", next_delivery.get("eta2", {})
)
if "eta2" in next_delivery:
del next_delivery["eta2"]
# Determine the total price by adding up the total price of all sub-orders
total_price = 0
for order in last_order.get("orders", []):
total_price += order.get("total_price", 0)
# Sanitise the object
last_order["total_price"] = total_price
last_order.setdefault("delivery_time", {})
if "eta2" in last_order:
del last_order["eta2"]
# Make a copy because some references are local
return last_order
# Make sure delivery_time is a dict
last_order.setdefault("delivery_time", {})
return next_delivery, last_order
@callback
def _update_auth_token(self):

112
tests/components/picnic/test_sensor.py
View File

@ -238,16 +238,6 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
cls=SensorDeviceClass.TIMESTAMP,
)
self._assert_sensor("sensor.picnic_last_order_status", "COMPLETED")
self._assert_sensor(
"sensor.picnic_last_order_eta_start",
"2021-02-26T19:54:00+00:00",
cls=SensorDeviceClass.TIMESTAMP,
)
self._assert_sensor(
"sensor.picnic_last_order_eta_end",
"2021-02-26T20:14:00+00:00",
cls=SensorDeviceClass.TIMESTAMP,
)
self._assert_sensor(
"sensor.picnic_last_order_max_order_time",
"2021-02-25T21:00:00+00:00",
@ -261,6 +251,26 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
self._assert_sensor(
"sensor.picnic_last_order_total_price", "41.33", unit=CURRENCY_EURO
)
self._assert_sensor(
"sensor.picnic_next_delivery_eta_start",
"unknown",
cls=SensorDeviceClass.TIMESTAMP,
)
self._assert_sensor(
"sensor.picnic_next_delivery_eta_end",
"unknown",
cls=SensorDeviceClass.TIMESTAMP,
)
self._assert_sensor(
"sensor.picnic_next_delivery_slot_start",
"unknown",
cls=SensorDeviceClass.TIMESTAMP,
)
self._assert_sensor(
"sensor.picnic_next_delivery_slot_end",
"unknown",
cls=SensorDeviceClass.TIMESTAMP,
)
async def test_sensors_setup_disabled_by_default(self):
"""Test that some sensors are disabled by default."""
@ -271,6 +281,8 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
self._assert_sensor("sensor.picnic_last_order_slot_end", disabled=True)
self._assert_sensor("sensor.picnic_last_order_status", disabled=True)
self._assert_sensor("sensor.picnic_last_order_total_price", disabled=True)
self._assert_sensor("sensor.picnic_next_delivery_slot_start", disabled=True)
self._assert_sensor("sensor.picnic_next_delivery_slot_end", disabled=True)
async def test_sensors_no_selected_time_slot(self):
"""Test sensor states with no explicit selected time slot."""
@ -295,11 +307,12 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
"sensor.picnic_selected_slot_min_order_value", STATE_UNKNOWN
)
async def test_sensors_last_order_in_future(self):
async def test_next_delivery_sensors(self):
"""Test sensor states when last order is not yet delivered."""
# Adjust default delivery response
delivery_response = copy.deepcopy(DEFAULT_DELIVERY_RESPONSE)
del delivery_response["delivery_time"]
delivery_response["status"] = "CURRENT"
# Set mock responses
self.picnic_mock().get_user.return_value = copy.deepcopy(DEFAULT_USER_RESPONSE)
@ -311,10 +324,16 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
# Assert delivery time is not available, but eta is
self._assert_sensor("sensor.picnic_last_order_delivery_time", STATE_UNKNOWN)
self._assert_sensor(
"sensor.picnic_last_order_eta_start", "2021-02-26T19:54:00+00:00"
"sensor.picnic_next_delivery_eta_start", "2021-02-26T19:54:00+00:00"
)
self._assert_sensor(
"sensor.picnic_last_order_eta_end", "2021-02-26T20:14:00+00:00"
"sensor.picnic_next_delivery_eta_end", "2021-02-26T20:14:00+00:00"
)
self._assert_sensor(
"sensor.picnic_next_delivery_slot_start", "2021-02-26T19:15:00+00:00"
)
self._assert_sensor(
"sensor.picnic_next_delivery_slot_end", "2021-02-26T20:15:00+00:00"
)
async def test_sensors_eta_date_malformed(self):
@ -329,12 +348,13 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
}
delivery_response = copy.deepcopy(DEFAULT_DELIVERY_RESPONSE)
delivery_response["eta2"] = eta_dates
delivery_response["status"] = "CURRENT"
self.picnic_mock().get_deliveries.return_value = [delivery_response]
await self._coordinator.async_refresh()
# Assert eta times are not available due to malformed date strings
self._assert_sensor("sensor.picnic_last_order_eta_start", STATE_UNKNOWN)
self._assert_sensor("sensor.picnic_last_order_eta_end", STATE_UNKNOWN)
self._assert_sensor("sensor.picnic_next_delivery_eta_start", STATE_UNKNOWN)
self._assert_sensor("sensor.picnic_next_delivery_eta_end", STATE_UNKNOWN)
async def test_sensors_use_detailed_eta_if_available(self):
"""Test sensor states when last order is not yet delivered."""
@ -344,6 +364,7 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
# Provide a delivery position response with different ETA and remove delivery time from response
delivery_response = copy.deepcopy(DEFAULT_DELIVERY_RESPONSE)
del delivery_response["delivery_time"]
delivery_response["status"] = "CURRENT"
self.picnic_mock().get_deliveries.return_value = [delivery_response]
self.picnic_mock().get_delivery_position.return_value = {
"eta_window": {
@ -358,10 +379,10 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
delivery_response["delivery_id"]
)
self._assert_sensor(
"sensor.picnic_last_order_eta_start", "2021-03-05T10:19:20+00:00"
"sensor.picnic_next_delivery_eta_start", "2021-03-05T10:19:20+00:00"
)
self._assert_sensor(
"sensor.picnic_last_order_eta_end", "2021-03-05T10:39:20+00:00"
"sensor.picnic_next_delivery_eta_end", "2021-03-05T10:39:20+00:00"
)
async def test_sensors_no_data(self):
@ -387,12 +408,12 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
self._assert_sensor(
"sensor.picnic_selected_slot_min_order_value", STATE_UNAVAILABLE
)
self._assert_sensor("sensor.picnic_last_order_eta_start", STATE_UNAVAILABLE)
self._assert_sensor("sensor.picnic_last_order_eta_end", STATE_UNAVAILABLE)
self._assert_sensor(
"sensor.picnic_last_order_max_order_time", STATE_UNAVAILABLE
)
self._assert_sensor("sensor.picnic_last_order_delivery_time", STATE_UNAVAILABLE)
self._assert_sensor("sensor.picnic_next_delivery_eta_start", STATE_UNAVAILABLE)
self._assert_sensor("sensor.picnic_next_delivery_eta_end", STATE_UNAVAILABLE)
async def test_sensors_malformed_delivery_data(self):
"""Test sensor states when the delivery api returns not a list."""
@ -405,10 +426,10 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
# Assert all last-order sensors have STATE_UNAVAILABLE because the delivery info fetch failed
assert self._coordinator.last_update_success is True
self._assert_sensor("sensor.picnic_last_order_eta_start", STATE_UNKNOWN)
self._assert_sensor("sensor.picnic_last_order_eta_end", STATE_UNKNOWN)
self._assert_sensor("sensor.picnic_last_order_max_order_time", STATE_UNKNOWN)
self._assert_sensor("sensor.picnic_last_order_delivery_time", STATE_UNKNOWN)
self._assert_sensor("sensor.picnic_next_delivery_eta_start", STATE_UNKNOWN)
self._assert_sensor("sensor.picnic_next_delivery_eta_end", STATE_UNKNOWN)
async def test_sensors_malformed_response(self):
"""Test coordinator update fails when API yields ValueError."""
@ -423,6 +444,55 @@ class TestPicnicSensor(unittest.IsolatedAsyncioTestCase):
# Assert coordinator update failed
assert self._coordinator.last_update_success is False
async def test_multiple_active_orders(self):
"""Test that the sensors get the right values when there are multiple active orders."""
# Create 2 undelivered orders
undelivered_order = copy.deepcopy(DEFAULT_DELIVERY_RESPONSE)
del undelivered_order["delivery_time"]
undelivered_order["status"] = "CURRENT"
undelivered_order["slot"]["window_start"] = "2022-03-01T09:15:00.000+01:00"
undelivered_order["slot"]["window_end"] = "2022-03-01T10:15:00.000+01:00"
undelivered_order["eta2"]["start"] = "2022-03-01T09:30:00.000+01:00"
undelivered_order["eta2"]["end"] = "2022-03-01T09:45:00.000+01:00"
undelivered_order_2 = copy.deepcopy(undelivered_order)
undelivered_order_2["slot"]["window_start"] = "2022-03-08T13:15:00.000+01:00"
undelivered_order_2["slot"]["window_end"] = "2022-03-08T14:15:00.000+01:00"
undelivered_order_2["eta2"]["start"] = "2022-03-08T13:30:00.000+01:00"
undelivered_order_2["eta2"]["end"] = "2022-03-08T13:45:00.000+01:00"
deliveries_response = [
undelivered_order_2,
undelivered_order,
copy.deepcopy(DEFAULT_DELIVERY_RESPONSE),
]
# Set mock responses
self.picnic_mock().get_user.return_value = copy.deepcopy(DEFAULT_USER_RESPONSE)
self.picnic_mock().get_cart.return_value = copy.deepcopy(DEFAULT_CART_RESPONSE)
self.picnic_mock().get_deliveries.return_value = deliveries_response
self.picnic_mock().get_delivery_position.return_value = {}
await self._setup_platform()
self._assert_sensor(
"sensor.picnic_last_order_slot_start", "2022-03-08T12:15:00+00:00"
)
self._assert_sensor(
"sensor.picnic_last_order_slot_end", "2022-03-08T13:15:00+00:00"
)
self._assert_sensor(
"sensor.picnic_next_delivery_slot_start", "2022-03-01T08:15:00+00:00"
)
self._assert_sensor(
"sensor.picnic_next_delivery_slot_end", "2022-03-01T09:15:00+00:00"
)
self._assert_sensor(
"sensor.picnic_next_delivery_eta_start", "2022-03-01T08:30:00+00:00"
)
self._assert_sensor(
"sensor.picnic_next_delivery_eta_end", "2022-03-01T08:45:00+00:00"
)
async def test_device_registry_entry(self):
"""Test if device registry entry is populated correctly."""
# Setup platform and default mock responses