Add support for journal events on a timeline.

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
+    )

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
 
 

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,
+        )
+    )

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"},
+        ),
+    ]