Refactor vulcan integration

homeassistant/components/vulcan/calendar.py

@@ -1,21 +1,25 @@
 """Support for Vulcan Calendar platform."""
-import copy
+from __future__ import annotations
+
 from datetime import date, datetime, timedelta
 import logging
 
 from aiohttp import ClientConnectorError
 from vulcan._utils import VulcanAPIException
 
-from homeassistant.components.calendar import ENTITY_ID_FORMAT, CalendarEventDevice
+from homeassistant.components.calendar import (
+    ENTITY_ID_FORMAT,
+    CalendarEntity,
+    CalendarEvent,
+)
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_SCAN_INTERVAL
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryAuthFailed
 from homeassistant.helpers.device_registry import DeviceEntryType
 from homeassistant.helpers.entity import generate_entity_id
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
-from homeassistant.helpers.template import DATE_STR_FORMAT
-from homeassistant.util import Throttle, dt
+from homeassistant.util import Throttle
 
 from . import DOMAIN
 from .const import DEFAULT_SCAN_INTERVAL
@@ -29,8 +33,8 @@ async def async_setup_entry(
     config_entry: ConfigEntry,
     async_add_entities: AddEntitiesCallback,
 ) -> None:
-    """Set up the calendar platform for event devices."""
-    VulcanCalendarData.MIN_TIME_BETWEEN_UPDATES = timedelta(
+    """Set up the calendar platform for entity."""
+    VulcanCalendarEntity.MIN_TIME_BETWEEN_UPDATES = timedelta(
         minutes=config_entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
     )
     client = hass.data[DOMAIN][config_entry.entry_id]
@@ -42,7 +46,7 @@ async def async_setup_entry(
     }
     async_add_entities(
         [
-            VulcanCalendarEventDevice(
+            VulcanCalendarEntity(
                 client,
                 data,
                 generate_entity_id(
@@ -55,20 +59,17 @@ async def async_setup_entry(
     )
 
 
-class VulcanCalendarEventDevice(CalendarEventDevice):
-    """A calendar event device."""
+class VulcanCalendarEntity(CalendarEntity):
+    """A calendar entity."""
 
-    def __init__(self, client, data, entity_id):
-        """Create the Calendar event device."""
+    def __init__(self, client, data, entity_id) -> None:
+        """Create the Calendar entity."""
         self.student_info = data["student_info"]
-        self.data = VulcanCalendarData(
-            client,
-            self.student_info,
-            self.hass,
-        )
-        self._event = None
+        self._event: CalendarEvent | None = None
+        self.client = client
         self.entity_id = entity_id
         self._unique_id = f"vulcan_calendar_{self.student_info['id']}"
+        self.service_available = True
 
         if data["students_number"] == 1:
             self._attr_name = "Vulcan calendar"
@@ -86,49 +87,14 @@ def __init__(self, client, data, entity_id):
             "configuration_url": f"https://uonetplus.vulcan.net.pl/{self.student_info['symbol']}",
         }
 
+    MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=DEFAULT_SCAN_INTERVAL)
+
     @property
-    def event(self):
+    def event(self) -> CalendarEvent | None:
         """Return the next upcoming event."""
         return self._event
 
-    async def async_get_events(self, hass, start_date, end_date):
-        """Get all events in a specific time frame."""
-        return await self.data.async_get_events(hass, start_date, end_date)
-
-    async def async_update(self):
-        """Update event data."""
-        await self.data.async_update()
-        event = copy.deepcopy(self.data.event)
-        if event is None:
-            self._event = event
-            return
-        event["start"] = {
-            "dateTime": datetime.combine(event["date"], event["time"].from_)
-            .astimezone(dt.DEFAULT_TIME_ZONE)
-            .isoformat()
-        }
-        event["end"] = {
-            "dateTime": datetime.combine(event["date"], event["time"].to)
-            .astimezone(dt.DEFAULT_TIME_ZONE)
-            .isoformat()
-        }
-        self._event = event
-
-
-class VulcanCalendarData:
-    """Class to utilize calendar service object to get next event."""
-
-    MIN_TIME_BETWEEN_UPDATES = timedelta(minutes=DEFAULT_SCAN_INTERVAL)
-
-    def __init__(self, client, student_info, hass):
-        """Set up how we are going to search the Vulcan calendar."""
-        self.client = client
-        self.event = None
-        self.hass = hass
-        self.student_info = student_info
-        self._available = True
-
-    async def async_get_events(self, hass, start_date, end_date):
+    async def async_get_events(self, hass, start_date, end_date) -> list[CalendarEvent]:
         """Get all events in a specific time frame."""
         try:
             events = await get_lessons(
@@ -145,53 +111,44 @@ async def async_get_events(self, hass, start_date, end_date):
             _LOGGER.error("An API error has occurred: %s", err)
             events = []
         except ClientConnectorError as err:
-            if self._available:
+            if self.service_available:
                 _LOGGER.warning(
                     "Connection error - please check your internet connection: %s", err
                 )
             events = []
 
         event_list = []
         for item in events:
-            event = {
-                "uid": item["id"],
-                "start": {
-                    "dateTime": datetime.combine(
-                        item["date"], item["time"].from_
-                    ).strftime(DATE_STR_FORMAT)
-                },
-                "end": {
-                    "dateTime": datetime.combine(
-                        item["date"], item["time"].to
-                    ).strftime(DATE_STR_FORMAT)
-                },
-                "summary": item["lesson"],
-                "location": item["room"],
-                "description": item["teacher"],
-            }
+            event = CalendarEvent(
+                start=datetime.combine(item["date"], item["time"].from_),
+                end=datetime.combine(item["date"], item["time"].to),
+                summary=item["lesson"],
+                location=item["room"],
+                description=item["teacher"],
+            )
 
             event_list.append(event)
 
         return event_list
 
     @Throttle(MIN_TIME_BETWEEN_UPDATES)
-    async def async_update(self):
+    async def async_update(self) -> None:
         """Get the latest data."""
 
         try:
             events = await get_lessons(self.client)
 
-            if not self._available:
+            if not self.service_available:
                 _LOGGER.info("Restored connection with API")
-                self._available = True
+                self.service_available = True
 
             if events == []:
                 events = await get_lessons(
                     self.client,
                     date_to=date.today() + timedelta(days=7),
                 )
                 if events == []:
-                    self.event = None
+                    self._event = None
                     return
         except VulcanAPIException as err:
             if str(err) == "The certificate is not authorized.":
@@ -202,11 +159,11 @@ async def async_update(self):
             _LOGGER.error("An API error has occurred: %s", err)
             return
         except ClientConnectorError as err:
-            if self._available:
+            if self.service_available:
                 _LOGGER.warning(
                     "Connection error - please check your internet connection: %s", err
                 )
-                self._available = False
+                self.service_available = False
             return
 
         new_event = min(
@@ -216,11 +173,10 @@ async def async_update(self):
                 abs(datetime.combine(d["date"], d["time"].to) - datetime.now()),
             ),
         )
-        self.event = {
-            "uid": new_event["id"],
-            "date": new_event["date"],
-            "time": new_event["time"],
-            "summary": new_event["lesson"],
-            "location": new_event["room"],
-            "description": new_event["teacher"],
-        }
+        self._event = CalendarEvent(
+            start=datetime.combine(new_event["date"], new_event["time"].from_),
+            end=datetime.combine(new_event["date"], new_event["time"].to),
+            summary=new_event["lesson"],
+            location=new_event["room"],
+            description=new_event["teacher"],
+        )