diff --git a/ical/journal.py b/ical/journal.py index e5fc7d9..09ddcb4 100644 --- a/ical/journal.py +++ b/ical/journal.py @@ -7,6 +7,7 @@ import datetime import enum import logging +from collections.abc import Iterable from typing import Any, Optional, Union try: @@ -16,11 +17,28 @@ from .component import ComponentModel, validate_until_dtstart, validate_recurrence_dates from .parsing.property import ParsedProperty -from .types import CalAddress, Classification, Recur, RecurrenceId, RequestStatus, Uri, RelatedTo -from .util import dtstamp_factory, normalize_datetime, uid_factory +from .types import ( + CalAddress, + Classification, + Recur, + RecurrenceId, + RequestStatus, + Uri, + RelatedTo, +) +from .exceptions import CalendarParseError +from .util import dtstamp_factory, normalize_datetime, uid_factory, local_timezone +from .iter import RulesetIterable +from .timespan import Timespan + _LOGGER = logging.getLogger(__name__) +__all__ = ["Journal", "JournalStatus"] + +_ONE_HOUR = datetime.timedelta(hours=1) +_ONE_DAY = datetime.timedelta(days=1) + class JournalStatus(str, enum.Enum): """Status or confirmation of the journal entry.""" @@ -97,5 +115,59 @@ def start_datetime(self) -> datetime.datetime: """Return the events start as a datetime.""" return normalize_datetime(self.start).astimezone(tz=datetime.timezone.utc) + @property + def computed_duration(self) -> datetime.timedelta: + """Return the event duration.""" + if isinstance(self.dtstart, datetime.datetime): + return _ONE_HOUR + return _ONE_DAY + + @property + def timespan(self) -> Timespan: + """Return a timespan representing the item start and due date.""" + return self.timespan_of(local_timezone()) + + def timespan_of(self, tzinfo: datetime.tzinfo) -> Timespan: + """Return a timespan representing the item start and due date.""" + dtstart = normalize_datetime(self.dtstart, tzinfo) or datetime.datetime.now( + tz=tzinfo + ) + return Timespan.of(dtstart, dtstart + self.computed_duration, tzinfo) + + @property + def recurring(self) -> bool: + """Return true if this Todo is recurring. + + A recurring event is typically evaluated specially on the list. The + data model has a single todo, but the timeline evaluates the recurrence + to expand and copy the the event to multiple places on the timeline + using `as_rrule`. + """ + if self.rrule or self.rdate: + return True + return False + + def as_rrule(self) -> Iterable[datetime.datetime | datetime.date] | None: + """Return an iterable containing the occurrences of a recurring todo. + + A recurring todo is typically evaluated specially on the todo list. The + data model has a single todo item, but the timeline evaluates the recurrence + to expand and copy the the item to multiple places on the timeline. + + This is only valid for events where `recurring` is True. + """ + if not self.rrule and not self.rdate: + return None + if not self.start: + raise CalendarParseError("Event must have a start date to be recurring") + return RulesetIterable( + self.start, + [self.rrule.as_rrule(self.start)] if self.rrule else [], + self.rdate, + [], + ) + _validate_until_dtstart = root_validator(allow_reuse=True)(validate_until_dtstart) - _validate_recurrence_dates = root_validator(allow_reuse=True)(validate_recurrence_dates) + _validate_recurrence_dates = root_validator(allow_reuse=True)( + validate_recurrence_dates + ) diff --git a/ical/recur_adapter.py b/ical/recur_adapter.py index 2976038..538f59a 100644 --- a/ical/recur_adapter.py +++ b/ical/recur_adapter.py @@ -20,10 +20,12 @@ from .types.recur import RecurrenceId from .event import Event from .todo import Todo +from .journal import Journal +from .freebusy import FreeBusy from .timespan import Timespan -ItemType = TypeVar("ItemType", bound="Event | Todo") +ItemType = TypeVar("ItemType", bound="Event | Todo | Journal") _DateOrDatetime = datetime.datetime | datetime.date diff --git a/ical/timeline.py b/ical/timeline.py index 8491a3f..bbf99ed 100644 --- a/ical/timeline.py +++ b/ical/timeline.py @@ -9,87 +9,37 @@ import datetime from collections.abc import Iterable, Iterator +from typing import TypeVar, Generic, Protocol from .event import Event from .iter import ( SortableItemTimeline, SpanOrderedItem, ) -from .recur_adapter import merge_and_expand_items +from .recur_adapter import merge_and_expand_items, ItemType -__all__ = ["Timeline"] +__all__ = ["Timeline", "generic_timeline"] - -class Timeline(SortableItemTimeline[Event]): - """A set of events on a calendar. - - A timeline is typically created from a `ics.calendar.Calendar` and is - typically not instantiated directly. - """ - - def __init__(self, iterable: Iterable[SpanOrderedItem[Event]]) -> None: - super().__init__(iterable) - - def __iter__(self) -> Iterator[Event]: - """Return an iterator as a traversal over events in chronological order.""" - return super().__iter__() - - def included( - self, - start: datetime.date | datetime.datetime, - end: datetime.date | datetime.datetime, - ) -> Iterator[Event]: - """Return an iterator for all events active during the timespan. - - The end date is exclusive. - """ - return super().included(start, end) - - def overlapping( - self, - start: datetime.date | datetime.datetime, - end: datetime.date | datetime.datetime, - ) -> Iterator[Event]: - """Return an iterator containing events active during the timespan. - - The end date is exclusive. - """ - return super().overlapping(start, end) - - def start_after( - self, - instant: datetime.datetime | datetime.date, - ) -> Iterator[Event]: - """Return an iterator containing events starting after the specified time.""" - return super().start_after(instant) - - def active_after( - self, - instant: datetime.datetime | datetime.date, - ) -> Iterator[Event]: - """Return an iterator containing events active after the specified time.""" - return super().active_after(instant) - - def at_instant( - self, - instant: datetime.date | datetime.datetime, - ) -> Iterator[Event]: # pylint: disable - """Return an iterator containing events starting after the specified time.""" - return super().at_instant(instant) - - def on_date(self, day: datetime.date) -> Iterator[Event]: # pylint: disable - """Return an iterator containing all events active on the specified day.""" - return super().on_date(day) - - def today(self) -> Iterator[Event]: - """Return an iterator containing all events active on the specified day.""" - return super().today() - - def now(self, tz: datetime.tzinfo | None = None) -> Iterator[Event]: - """Return an iterator containing all events active on the specified day.""" - return super().now(tz) +Timeline = SortableItemTimeline[Event] def calendar_timeline(events: list[Event], tzinfo: datetime.tzinfo) -> Timeline: """Create a timeline for events on a calendar, including recurrence.""" return Timeline(merge_and_expand_items(events, tzinfo)) + + +def generic_timeline( + items: list[ItemType], tzinfo: datetime.tzinfo +) -> SortableItemTimeline[ItemType]: + """Return a timeline view of events on the calendar. + + All events are returned as if the attendee is viewing from the + specified timezone. For example, this affects the order that All Day + events are returned. + """ + return SortableItemTimeline( + merge_and_expand_items( + items, + tzinfo, + ) + ) diff --git a/tests/test_timeline.py b/tests/test_timeline.py index c3ad856..30f7aa6 100644 --- a/tests/test_timeline.py +++ b/tests/test_timeline.py @@ -9,7 +9,9 @@ from ical.calendar import Calendar from ical.event import Event +from ical.journal import Journal from ical.types.recur import Recur +from ical.timeline import generic_timeline TZ = zoneinfo.ZoneInfo("America/Regina") @@ -53,3 +55,35 @@ def exhaust() -> int: result = benchmark(exhaust) assert result == num_events * num_instances + + +def test_journal_timeline() -> None: + """Test journal entries on a timeline.""" + + journal = Journal( + summary="Example", + start=datetime.date(2022, 8, 7), + rrule=Recur.from_rrule("FREQ=DAILY;COUNT=3"), + ) + + timeline = generic_timeline([journal], TZ) + assert list(timeline) == [ + Journal.copy(journal, update={"recurrence_id": "20220807"}), + Journal.copy( + journal, + update={"dtstart": datetime.date(2022, 8, 8), "recurrence_id": "20220808"}, + ), + Journal.copy( + journal, + update={"dtstart": datetime.date(2022, 8, 9), "recurrence_id": "20220809"}, + ), + ] + assert list( + timeline.overlapping(datetime.date(2022, 8, 7), datetime.date(2022, 8, 9)) + ) == [ + Journal.copy(journal, update={"recurrence_id": "20220807"}), + Journal.copy( + journal, + update={"dtstart": datetime.date(2022, 8, 8), "recurrence_id": "20220808"}, + ), + ]