Skip to content

Fix NoneType crash in premium sensors and cart service calls #66

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
TeTeHacko wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Fix NoneType crash in premium sensors and cart service calls #66

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 91 additions & 32 deletions custom_components/rohlikcz/binary_sensor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""Platform for binary sensor."""

from __future__ import annotations

from collections.abc import Mapping
Expand All @@ -10,37 +11,62 @@
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback

from .const import DOMAIN, ICON_REUSABLE, ICON_PARENTCLUB, ICON_PREMIUM, ICON_ORDER, ICON_TIMESLOT, ICON_CALENDAR_CHECK, \
ICON_CALENDAR_REMOVE
from .const import (
DOMAIN,
ICON_REUSABLE,
ICON_PARENTCLUB,
ICON_PREMIUM,
ICON_ORDER,
ICON_TIMESLOT,
ICON_CALENDAR_CHECK,
ICON_CALENDAR_REMOVE,
)
from .entity import BaseEntity
from .hub import RohlikAccount
from .utils import get_earliest_order


async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback
async_add_entities: AddEntitiesCallback,
) -> None:
"""Add sensors for passed config_entry in HA."""
rohlik_account: RohlikAccount = hass.data[DOMAIN][config_entry.entry_id] # type: ignore[Any]
async_add_entities([
IsReusableSensor(rohlik_account),
IsParentSensor(rohlik_account),
IsPremiumSensor(rohlik_account),
IsOrderedSensor(rohlik_account),
IsReservedSensor(rohlik_account),
IsExpressAvailable(rohlik_account)
])
async_add_entities(
[
IsReusableSensor(rohlik_account),
IsParentSensor(rohlik_account),
IsPremiumSensor(rohlik_account),
IsOrderedSensor(rohlik_account),
IsReservedSensor(rohlik_account),
IsExpressAvailable(rohlik_account),
]
)


class IsExpressAvailable(BaseEntity, BinarySensorEntity):
_attr_translation_key = "is_express_available"
_attr_should_poll = False

@property
def is_on(self) -> bool | None:
if not self._rohlik_account.data["next_delivery_slot"].get('data', {}).get('expressSlot', None):
if (
not self._rohlik_account.data["next_delivery_slot"]
.get("data", {})
.get("expressSlot", None)
):
return False
elif int(self._rohlik_account.data["next_delivery_slot"].get('data', {}).get('expressSlot', {}).get("timeSlotCapacityDTO", {}).get("totalFreeCapacityPercent", 0)) == 0:
elif (
int(
self._rohlik_account.data["next_delivery_slot"]
.get("data", {})
.get("expressSlot", {})
.get("timeSlotCapacityDTO", {})
.get("totalFreeCapacityPercent", 0)
)
== 0
):
return False
else:
return True
Expand Down Expand Up @@ -72,7 +98,12 @@ class IsReusableSensor(BaseEntity, BinarySensorEntity):

@property
def is_on(self) -> bool | None:
return self._rohlik_account.data.get('login', {}).get('data', {}).get('user', {}).get('reusablePackaging', False)
return (
self._rohlik_account.data.get("login", {})
.get("data", {})
.get("user", {})
.get("reusablePackaging", False)
)

@property
def icon(self) -> str:
Expand All @@ -98,7 +129,12 @@ class IsParentSensor(BaseEntity, BinarySensorEntity):

@property
def is_on(self) -> bool | None:
return self._rohlik_account.data.get('login', {}).get('data', {}).get('user', {}).get('parentsClub', False)
return (
self._rohlik_account.data.get("login", {})
.get("data", {})
.get("user", {})
.get("parentsClub", False)
)

@property
def icon(self) -> str:
Expand All @@ -122,21 +158,38 @@ class IsPremiumSensor(BaseEntity, BinarySensorEntity):

@property
def is_on(self) -> bool | None:
return self._rohlik_account.data.get('login', {}).get('data', {}).get('user', {}).get('premium', {}).get('active', False)
return (
self._rohlik_account.data.get("login", {})
.get("data", {})
.get("user", {})
.get("premium", {})
.get("active", False)
)

@property
def extra_state_attributes(self) -> dict | None:
premium_data = self._rohlik_account.data.get('login', {}).get('data', {}).get('user', {}).get('premium', {})
premium_data = (
self._rohlik_account.data.get("login", {})
.get("data", {})
.get("user", {})
.get("premium", {})
)
if premium_data:
return {
"type": premium_data.get('premiumMembershipType'),
"payment_type": premium_data.get('premiumType'),
"expiration_date": premium_data.get('recurrentPaymentDate'),
"remaining_days": premium_data.get('remainingDays'),
"start_date": premium_data.get('startDate'),
"end_date": premium_data.get('endDate'),
"remaining_orders_without_limit": premium_data.get('premiumLimits', {}).get('ordersWithoutPriceLimit', {}).get('remaining'),
"remaining_free_express": premium_data.get('premiumLimits', {}).get('freeExpressLimit', {}).get('remaining')
"type": premium_data.get("premiumMembershipType"),
"payment_type": premium_data.get("premiumType"),
"expiration_date": premium_data.get("recurrentPaymentDate"),
"remaining_days": premium_data.get("remainingDays"),
"start_date": premium_data.get("startDate"),
"end_date": premium_data.get("endDate"),
"remaining_orders_without_limit": (
premium_data.get("premiumLimits") or {}
)
.get("ordersWithoutPriceLimit", {})
.get("remaining"),
"remaining_free_express": (premium_data.get("premiumLimits") or {})
.get("freeExpressLimit", {})
.get("remaining"),
}
return None

Expand All @@ -162,16 +215,14 @@ class IsOrderedSensor(BaseEntity, BinarySensorEntity):
@property
def is_on(self) -> bool | None:
# Check if there's at least one order in the next_order list
return len(self._rohlik_account.data.get('next_order', [])) > 0
return len(self._rohlik_account.data.get("next_order", [])) > 0

@property
def extra_state_attributes(self) -> dict | None:
next_orders = self._rohlik_account.data.get('next_order', [])
next_orders = self._rohlik_account.data.get("next_order", [])
order = get_earliest_order(next_orders)
if order:
return {
"order_data": order
}
return {"order_data": order}
return None

@property
Expand All @@ -195,11 +246,19 @@ class IsReservedSensor(BaseEntity, BinarySensorEntity):

@property
def is_on(self) -> bool | None:
return self._rohlik_account.data.get('timeslot', {}).get('data', {}).get('active', False)
return (
self._rohlik_account.data.get("timeslot", {})
.get("data", {})
.get("active", False)
)

@property
def extra_state_attributes(self) -> dict | None:
timeslot_data = self._rohlik_account.data.get('timeslot', {}).get('data', {}).get('reservationDetail', {})
timeslot_data = (
self._rohlik_account.data.get("timeslot", {})
.get("data", {})
.get("reservationDetail", {})
)
if timeslot_data:
return timeslot_data
return None
Expand Down
48 changes: 33 additions & 15 deletions custom_components/rohlikcz/hub.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,12 @@ def has_address(self):

@property
def device_info(self) -> DeviceInfo:
""" Provides a device info. """
return {"identifiers": {(DOMAIN, self.data["login"]["data"]["user"]["id"])}, "name": self.data["login"]["data"]["user"]["name"], "manufacturer": "Rohlík.cz"}
"""Provides a device info."""
return {
"identifiers": {(DOMAIN, self.data["login"]["data"]["user"]["id"])},
"name": self.data["login"]["data"]["user"]["name"],
"manufacturer": "Rohlík.cz",
}

@property
def name(self) -> str:
Expand All @@ -45,10 +49,10 @@ def unique_id(self) -> str:

@property
def is_ordered(self) -> bool:
return len(self.data.get('next_order', [])) > 0
return len(self.data.get("next_order", [])) > 0

async def async_update(self) -> None:
""" Updates the data from API."""
"""Updates the data from API."""

self.data = await self._rohlik_api.get_data()

Expand All @@ -72,10 +76,11 @@ async def add_to_cart(self, product_id: int, quantity: int) -> Dict:
"""Add a product to the shopping cart."""
product_list = [{"product_id": product_id, "quantity": quantity}]
result = await self._rohlik_api.add_to_cart(product_list)
await self.async_update()
return result

async def search_product(self, product_name: str, limit: int = 10, favourite: bool = False) -> Optional[Dict[str, Any]]:
async def search_product(
self, product_name: str, limit: int = 10, favourite: bool = False
) -> Optional[Dict[str, Any]]:
"""Search for a product by name."""
result = await self._rohlik_api.search_product(product_name, limit, favourite)
return result
Expand All @@ -86,24 +91,37 @@ async def get_shopping_list(self, shopping_list_id: str) -> Dict[str, Any]:
return result

async def get_cart_content(self) -> Dict:
""" Retrieves cart content. """
"""Retrieves cart content."""
result = await self._rohlik_api.get_cart_content()
return result

async def search_and_add(self, product_name: str, quantity: int, favourite: bool = False) -> Dict | None:
""" Searches for product by name and adds to cart"""
async def search_and_add(
self, product_name: str, quantity: int, favourite: bool = False
) -> Dict | None:
"""Searches for product by name and adds to cart"""

searched_product = await self.search_product(product_name, limit = 5, favourite=favourite)
searched_product = await self.search_product(
product_name, limit=5, favourite=favourite
)

if searched_product:
await self.add_to_cart(searched_product["search_results"][0]["id"], quantity)
return {"success": True, "message": "", "added_to_cart": [searched_product["search_results"][0]]}
await self.add_to_cart(
searched_product["search_results"][0]["id"], quantity
)
return {
"success": True,
"message": "",
"added_to_cart": [searched_product["search_results"][0]],
}

else:
return {"success": False, "message": f'No product matched when searching for "{product_name}"{' in favourites' if favourite else ''}.', "added_to_cart": []}
return {
"success": False,
"message": f'No product matched when searching for "{product_name}"{" in favourites" if favourite else ""}.',
"added_to_cart": [],
}

async def delete_from_cart(self, order_field_id: str) -> Dict:
"""Delete a product from the shopping cart using orderFieldId."""
result = await self._rohlik_api.delete_from_cart(order_field_id)
await self.async_update() # Refresh data after deletion
return result
return result
Loading