Skip to content

Commit

Permalink
Merge pull request #27 from rine77/8-feature_request-school-timetable
Browse files Browse the repository at this point in the history
8 feature request school timetable
  • Loading branch information
rine77 authored Nov 24, 2024
2 parents ea2aaf0 + b0ba412 commit e626ccb
Show file tree
Hide file tree
Showing 6 changed files with 337 additions and 119 deletions.
130 changes: 79 additions & 51 deletions custom_components/homeassistantedupage/__init__.py
Original file line number Diff line number Diff line change
@@ -1,37 +1,43 @@
import logging
import asyncio
from datetime import timedelta
import datetime
from datetime import datetime, timedelta
from edupage_api.exceptions import BadCredentialsException, CaptchaException
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.const import Platform
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .homeassistant_edupage import Edupage
from edupage_api.classes import Class
from edupage_api.people import EduTeacher
from edupage_api.people import Gender
from edupage_api.classrooms import Classroom
from .const import DOMAIN

_LOGGER = logging.getLogger("custom_components.homeassistant_edupage")

async def async_setup(hass: HomeAssistant, config: dict) -> bool:
"""only ConfigEntry supported, no configuration.yaml yet"""
_LOGGER.info("INIT called async_setup")
_LOGGER.debug("INIT called async_setup")
return True

async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""initializin EduPage-integration and validate API-login"""
_LOGGER.info("INIT called async_setup_entry")
_LOGGER.debug("INIT called async_setup_entry")
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {}

username = entry.data["username"]
password = entry.data["password"]
subdomain = entry.data["subdomain"]
student_id = entry.data["student_id"]
edupage = Edupage(hass)
unique_id_sensorGrade = f"edupage_{username}_gradesensor"
coordinator = None

try:
login_success = await hass.async_add_executor_job(
edupage.login, username, password, subdomain
)
_LOGGER.info("INIT login_success")
_LOGGER.debug("INIT login_success")

except BadCredentialsException as e:
_LOGGER.error("INIT login failed: bad credentials. %s", e)
Expand All @@ -48,67 +54,89 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
fetch_lock = asyncio.Lock()

async def fetch_data():
"""Function to fetch grade and timetable data."""
_LOGGER.info("INIT called fetch_data")
"""Function to fetch timetable data for the selected student."""
_LOGGER.debug("INIT called fetch_data")

async with fetch_lock:
try:
# request classes
classes_data = await edupage.get_classes()
# _LOGGER.info("INIT classes count: " + str(len(classes_data)))

# request grades
grades_data = await edupage.get_grades()
# _LOGGER.info("INIT grade count: " + str(len(grades_data)))

# request user_id
userid = await edupage.get_user_id()
# _LOGGER.info("INIT user_id: "+str(userid))

# request all possible subjects
subjects_data = await edupage.get_subjects()
# _LOGGER.info("INIT subject count: " + str(len(subjects_data)))

# request all possible students
students_data = await edupage.get_students()
# _LOGGER.info("INIT students count: " + str(len(students_data)))

return {
"grades": grades_data,
# "timetable": timetable_data,
"user_id": userid,
"subjects": subjects_data
await edupage.login(username, password, subdomain)

students = await edupage.get_students()
student = next((s for s in students if s.person_id == student_id), None)
_LOGGER.debug("INIT Student: %s", student)

if not student:
_LOGGER.error("INIT No matching student found with ID: %s", student_id)
return {"timetable": {}}

_LOGGER.debug("INIT Found EduStudent: %s", vars(student))

grades = await edupage.get_grades()
subjects = await edupage.get_subjects()

timetable_data = {}
today = datetime.now().date()
for offset in range(14):
current_date = today + timedelta(days=offset)
timetable = await edupage.get_timetable(student, current_date)
if timetable:
_LOGGER.debug(f"Timetable for {current_date}: {timetable}")
timetable_data[current_date] = timetable
else:
_LOGGER.warning(f"INIT No timetable found for {current_date}")

return_data = {
"student": {"id": student.person_id, "name": student.name},
"grades": grades,
"subjects": subjects,
"timetable": timetable_data,
}
_LOGGER.debug(f"INIT Coordinator fetch_data returning: {return_data}")
return return_data

except Exception as e:
_LOGGER.error("INIT error fetching data: %s", e)
return False

coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name="EduPage Data",
update_method=fetch_data,
update_interval=timedelta(minutes=5),
)

# First data fetch
_LOGGER.error("INIT Failed to fetch timetable: %s", e)
return {"timetable": {}}

try:
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
name="Edupage",
update_method=fetch_data,
update_interval=timedelta(minutes=30),
)
hass.data[DOMAIN][entry.entry_id] = coordinator

await coordinator.async_config_entry_first_refresh()
_LOGGER.debug("INIT Coordinator successfully initialized")

except Exception as e:
_LOGGER.error(f"INIT Error during async_setup_entry: {e}")

if entry.entry_id in hass.data[DOMAIN]:
del hass.data[DOMAIN][entry.entry_id]

return False

await asyncio.sleep(1)
await coordinator.async_config_entry_first_refresh()
_LOGGER.debug(f"INIT Coordinator first fetch!")

# Save coordinator
hass.data.setdefault(DOMAIN, {})
hass.data[DOMAIN][entry.entry_id] = coordinator

# Forward platforms
await hass.config_entries.async_forward_entry_setups(entry, [Platform.SENSOR])
await hass.config_entries.async_forward_entry_setups(entry, ["calendar", "sensor"])
_LOGGER.debug(f"INIT forwarded")
_LOGGER.debug(f"INIT Coordinator data: {coordinator.data}")

return True


async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload ConfigEntry."""
_LOGGER.info("INIT called async_unload_entry")
unload_ok = await hass.config_entries.async_forward_entry_unload(entry, Platform.SENSOR)
_LOGGER.debug("INIT called async_unload_entry")
unload_ok = await hass.config_entries.async_forward_entry_unload(entry, ["calendar", "sensor"])
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

150 changes: 116 additions & 34 deletions custom_components/homeassistantedupage/calendar.py
Original file line number Diff line number Diff line change
@@ -1,48 +1,130 @@
import voluptuous as vol
import logging
from datetime import datetime, timedelta
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from datetime import datetime, timedelta, timezone
from homeassistant.components.calendar import CalendarEntity, CalendarEvent
from homeassistant.helpers.update_coordinator import UpdateFailed
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.core import HomeAssistant
from .homeassistant_edupage import Edupage
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from collections import defaultdict
from zoneinfo import ZoneInfo

_LOGGER = logging.getLogger("custom_components.homeassistant_edupage")
_LOGGER.debug("CALENDAR Edupage calendar.py is being loaded")

async def async_setup_entry(hass, entry, async_add_entities: AddEntitiesCallback) -> None:
"""Set up Edupage calendar entities."""
_LOGGER.debug("CALENDAR called async_setup_entry")

coordinator = hass.data[DOMAIN][entry.entry_id]

from homeassistant.components.calendar import CalendarEntity
edupage_calendar = EdupageCalendar(coordinator, entry.data)

_LOGGER = logging.getLogger(__name__)
async_add_entities([edupage_calendar])

async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback):
_LOGGER.debug("CALENDAR async_setup_entry finished.")

username = entry.data["username"]
password = entry.data["password"]
subdomain = entry.data["subdomain"]
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
_LOGGER.debug("CALENDAR added to hass")

edupage = Edupage(hass)
unique_id = f"edupage_{username}_calendar"
await hass.async_add_executor_job(edupage.login, username, password, subdomain)
if self.coordinator:
self.async_on_remove(
self.coordinator.async_add_listener(
self._handle_coordinator_update, None
)
)

async def async_update_data():
class EdupageCalendar(CoordinatorEntity, CalendarEntity):
"""Representation of an Edupage calendar entity."""

today = datetime.now().date()
try:
return await edupage.get_timetable(today)
except Exception as e:
_LOGGER.error(f"error updating data: {e}")
raise UpdateFailed(F"error updating data: {e}")
def __init__(self, coordinator, data):
super().__init__(coordinator)
self._data = data
self._events = []
self._attr_name = "Edupage Calendar"
_LOGGER.debug(f"CALENDAR Initialized EdupageCalendar with data: {data}")

async_add_entities([TimetableCalendar(edupage, unique_id)], True)
@property
def unique_id(self):
"""Return a unique ID for this calendar."""
student_id = self.coordinator.data.get("student", {}).get("id", "unknown")
return f"edupage_calendar_{student_id}"

class TimetableCalendar(CalendarEntity):
def __init__(self, edupage, unique_id):
self.edupage = edupage
self._attr_unique_id = unique_id

@property
def name(self):
"""return name of calendar"""
return self._name

async def async_get_events(self, today: datetime):
timetable = self.get_timetable(today)
# Konvertieren Sie 'timetable' in eine Liste von Ereignissen, die von dieser Methode zurückgegeben werden
return timetable #dict
"""Return the name of the calendar."""
student_name = self.coordinator.data.get("student", {}).get("name", "Unknown Student")
return f"Edupage - {student_name}"

@property
def extra_state_attributes(self):
"""Return the extra state attributes."""
return {
"unique_id": self.unique_id,
"other_info": "debug info"
}

@property
def available(self) -> bool:
"""Return True if the calendar is available."""
_LOGGER.debug("CALENDAR Checking availability of Edupage Calendar")
return True

@property
def event(self):
"""Return the next upcoming event or None if no event exists."""
local_tz = ZoneInfo(self.hass.config.time_zone)
now = datetime.now(local_tz)
return CalendarEvent(
start=now + timedelta(hours=1),
end=now + timedelta(hours=2),
summary="Next Dummy Event",
description="A placeholder event for debugging."
)

async def async_get_events(self, hass, start_date: datetime, end_date: datetime):
"""Return events in a specific date range."""
local_tz = ZoneInfo(self.hass.config.time_zone)
events = []

_LOGGER.debug(f"CALENDAR Fetching events between {start_date} and {end_date}")
timetable = self.coordinator.data.get("timetable", {})
_LOGGER.debug(f"CALENDAR Coordinator data: {self.coordinator.data}")
_LOGGER.debug(f"CALENDAR Fetched timetable data: {timetable}")

if not timetable:
_LOGGER.warning("CALENDAR Timetable data is missing.")
return events

current_date = start_date.date()
while current_date <= end_date.date():
day_timetable = timetable.get(current_date)
if day_timetable:
for lesson in day_timetable:

_LOGGER.debug(f"CALENDAR Lesson attributes: {vars(lesson)}")

room = "Unknown"
if lesson.classes and lesson.classes[0].homeroom:
room = lesson.classes[0].homeroom.name

teacher_names = [teacher.name for teacher in lesson.teachers]
teachers = ", ".join(teacher_names) if teacher_names else "Unknown Teacher"

start_time = datetime.combine(current_date, lesson.start_time).astimezone(local_tz)
end_time = datetime.combine(current_date, lesson.end_time).astimezone(local_tz)
events.append(
CalendarEvent(
start=start_time,
end=end_time,
summary=lesson.subject.name if lesson.subject else "Unknown Subject",
description=f"Room: {room}\nTeacher(s): {teachers}"
)
)
current_date += timedelta(days=1)

_LOGGER.debug(f"CALENDAR Fetched {len(events)} events from {start_date} to {end_date}")
return events
Loading

0 comments on commit e626ccb

Please sign in to comment.