mirror of
https://github.com/home-assistant/core.git
synced 2025-07-16 01:37:08 +00:00
Add additional type hints for calendar integration (#68660)
This commit is contained in:
parent
53110f8cb7
commit
f05a6826de
@ -1,11 +1,11 @@
|
|||||||
"""Support for Google Calendar event device sensors."""
|
"""Support for Google Calendar event device sensors."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from datetime import timedelta
|
import datetime
|
||||||
from http import HTTPStatus
|
from http import HTTPStatus
|
||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
from typing import cast, final
|
from typing import Any, cast, final
|
||||||
|
|
||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
|
|
||||||
@ -30,7 +30,7 @@ _LOGGER = logging.getLogger(__name__)
|
|||||||
|
|
||||||
DOMAIN = "calendar"
|
DOMAIN = "calendar"
|
||||||
ENTITY_ID_FORMAT = DOMAIN + ".{}"
|
ENTITY_ID_FORMAT = DOMAIN + ".{}"
|
||||||
SCAN_INTERVAL = timedelta(seconds=60)
|
SCAN_INTERVAL = datetime.timedelta(seconds=60)
|
||||||
|
|
||||||
|
|
||||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||||
@ -62,18 +62,22 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||||||
return await component.async_unload_entry(entry)
|
return await component.async_unload_entry(entry)
|
||||||
|
|
||||||
|
|
||||||
def get_date(date):
|
def get_date(date: dict[str, Any]) -> datetime.datetime:
|
||||||
"""Get the dateTime from date or dateTime as a local."""
|
"""Get the dateTime from date or dateTime as a local."""
|
||||||
if "date" in date:
|
if "date" in date:
|
||||||
|
parsed_date = dt.parse_date(date["date"])
|
||||||
|
assert parsed_date
|
||||||
return dt.start_of_local_day(
|
return dt.start_of_local_day(
|
||||||
dt.dt.datetime.combine(dt.parse_date(date["date"]), dt.dt.time.min)
|
datetime.datetime.combine(parsed_date, datetime.time.min)
|
||||||
)
|
)
|
||||||
return dt.as_local(dt.parse_datetime(date["dateTime"]))
|
parsed_datetime = dt.parse_datetime(date["dateTime"])
|
||||||
|
assert parsed_datetime
|
||||||
|
return dt.as_local(parsed_datetime)
|
||||||
|
|
||||||
|
|
||||||
def normalize_event(event):
|
def normalize_event(event: dict[str, Any]) -> dict[str, Any]:
|
||||||
"""Normalize a calendar event."""
|
"""Normalize a calendar event."""
|
||||||
normalized_event = {}
|
normalized_event: dict[str, Any] = {}
|
||||||
|
|
||||||
start = event.get("start")
|
start = event.get("start")
|
||||||
end = event.get("end")
|
end = event.get("end")
|
||||||
@ -97,7 +101,7 @@ def normalize_event(event):
|
|||||||
return normalized_event
|
return normalized_event
|
||||||
|
|
||||||
|
|
||||||
def calculate_offset(event, offset):
|
def calculate_offset(event: dict[str, Any], offset: str) -> dict[str, Any]:
|
||||||
"""Calculate event offset.
|
"""Calculate event offset.
|
||||||
|
|
||||||
Return the updated event with the offset_time included.
|
Return the updated event with the offset_time included.
|
||||||
@ -119,32 +123,33 @@ def calculate_offset(event, offset):
|
|||||||
summary = (summary[: search.start()] + summary[search.end() :]).strip()
|
summary = (summary[: search.start()] + summary[search.end() :]).strip()
|
||||||
event["summary"] = summary
|
event["summary"] = summary
|
||||||
else:
|
else:
|
||||||
offset_time = dt.dt.timedelta() # default it
|
offset_time = datetime.timedelta() # default it
|
||||||
|
|
||||||
event["offset_time"] = offset_time
|
event["offset_time"] = offset_time
|
||||||
return event
|
return event
|
||||||
|
|
||||||
|
|
||||||
def is_offset_reached(event):
|
def is_offset_reached(event: dict[str, Any]) -> bool:
|
||||||
"""Have we reached the offset time specified in the event title."""
|
"""Have we reached the offset time specified in the event title."""
|
||||||
start = get_date(event["start"])
|
start = get_date(event["start"])
|
||||||
if start is None or event["offset_time"] == dt.dt.timedelta():
|
offset_time: datetime.timedelta = event["offset_time"]
|
||||||
|
if start is None or offset_time == datetime.timedelta():
|
||||||
return False
|
return False
|
||||||
|
|
||||||
return start + event["offset_time"] <= dt.now(start.tzinfo)
|
return start + offset_time <= dt.now(start.tzinfo)
|
||||||
|
|
||||||
|
|
||||||
class CalendarEventDevice(Entity):
|
class CalendarEventDevice(Entity):
|
||||||
"""Base class for calendar event entities."""
|
"""Base class for calendar event entities."""
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def event(self):
|
def event(self) -> dict[str, Any] | None:
|
||||||
"""Return the next upcoming event."""
|
"""Return the next upcoming event."""
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
|
|
||||||
@final
|
@final
|
||||||
@property
|
@property
|
||||||
def state_attributes(self):
|
def state_attributes(self) -> dict[str, Any] | None:
|
||||||
"""Return the entity state attributes."""
|
"""Return the entity state attributes."""
|
||||||
if (event := self.event) is None:
|
if (event := self.event) is None:
|
||||||
return None
|
return None
|
||||||
@ -160,7 +165,7 @@ class CalendarEventDevice(Entity):
|
|||||||
}
|
}
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def state(self):
|
def state(self) -> str | None:
|
||||||
"""Return the state of the calendar event."""
|
"""Return the state of the calendar event."""
|
||||||
if (event := self.event) is None:
|
if (event := self.event) is None:
|
||||||
return STATE_OFF
|
return STATE_OFF
|
||||||
@ -179,7 +184,12 @@ class CalendarEventDevice(Entity):
|
|||||||
|
|
||||||
return STATE_OFF
|
return STATE_OFF
|
||||||
|
|
||||||
async def async_get_events(self, hass, start_date, end_date):
|
async def async_get_events(
|
||||||
|
self,
|
||||||
|
hass: HomeAssistant,
|
||||||
|
start_date: datetime.datetime,
|
||||||
|
end_date: datetime.datetime,
|
||||||
|
) -> list[dict[str, Any]]:
|
||||||
"""Return calendar events within a datetime range."""
|
"""Return calendar events within a datetime range."""
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
|
|
||||||
@ -194,18 +204,21 @@ class CalendarEventView(http.HomeAssistantView):
|
|||||||
"""Initialize calendar view."""
|
"""Initialize calendar view."""
|
||||||
self.component = component
|
self.component = component
|
||||||
|
|
||||||
async def get(self, request, entity_id):
|
async def get(self, request: web.Request, entity_id: str) -> web.Response:
|
||||||
"""Return calendar events."""
|
"""Return calendar events."""
|
||||||
entity = self.component.get_entity(entity_id)
|
entity = self.component.get_entity(entity_id)
|
||||||
start = request.query.get("start")
|
start = request.query.get("start")
|
||||||
end = request.query.get("end")
|
end = request.query.get("end")
|
||||||
if None in (start, end, entity):
|
if start is None or end is None or entity is None:
|
||||||
return web.Response(status=HTTPStatus.BAD_REQUEST)
|
return web.Response(status=HTTPStatus.BAD_REQUEST)
|
||||||
|
assert isinstance(entity, CalendarEventDevice)
|
||||||
try:
|
try:
|
||||||
start_date = dt.parse_datetime(start)
|
start_date = dt.parse_datetime(start)
|
||||||
end_date = dt.parse_datetime(end)
|
end_date = dt.parse_datetime(end)
|
||||||
except (ValueError, AttributeError):
|
except (ValueError, AttributeError):
|
||||||
return web.Response(status=HTTPStatus.BAD_REQUEST)
|
return web.Response(status=HTTPStatus.BAD_REQUEST)
|
||||||
|
if start_date is None or end_date is None:
|
||||||
|
return web.Response(status=HTTPStatus.BAD_REQUEST)
|
||||||
event_list = await entity.async_get_events(
|
event_list = await entity.async_get_events(
|
||||||
request.app["hass"], start_date, end_date
|
request.app["hass"], start_date, end_date
|
||||||
)
|
)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user