feat: Add stringification to Schedule and include evaluation on rules.

actual/rules.py

@@ -129,6 +129,8 @@ def condition_evaluation(
             # Actual uses two days as reference
             # https://github.com/actualbudget/actual/blob/98a7aac73667241da350169e55edd2fc16a6687f/packages/loot-core/src/server/accounts/rules.ts#L302-L304
             interval = datetime.timedelta(days=2)
+            if isinstance(self_value, Schedule):
+                return self_value.is_approx(true_value, interval)
         else:
             # Actual uses 7.5% of the value as threshold
             # https://github.com/actualbudget/actual/blob/243703b2f70532ec1acbd3088dda879b5d07a5b3/packages/loot-core/src/shared/rules.ts#L261-L263

actual/schedules.py

@@ -1,5 +1,6 @@
 import datetime
 import enum
+import typing
 
 import pydantic
 from dateutil.rrule import (
@@ -14,6 +15,22 @@
 )
 
 
+def date_to_datetime(date: typing.Optional[datetime.date]) -> typing.Optional[datetime.datetime]:
+    """Converts one object from date to datetime object. The reverse is possible directly by calling datetime.date()."""
+    if date is None:
+        return None
+    return datetime.datetime.combine(date, datetime.time.min)
+
+
+def day_to_ordinal(day: int) -> str:
+    """Converts an integer day to an ordinal number, i.e. 1 -> 1st, 32 -> 32nd"""
+    if 11 <= (day % 100) <= 13:
+        suffix = "th"
+    else:
+        suffix = ["th", "st", "nd", "rd", "th"][min(day % 10, 4)]
+    return f"{day}{suffix}"
+
+
 class EndMode(enum.Enum):
     AFTER_N_OCCURRENCES = "after_n_occurrences"
     ON_DATE = "on_date"
@@ -26,7 +43,7 @@ class Frequency(enum.Enum):
     MONTHLY = "monthly"
     YEARLY = "yearly"
 
-    def as_dateutil(self):
+    def as_dateutil(self) -> int:
         frequency_map = {"YEARLY": YEARLY, "MONTHLY": MONTHLY, "WEEKLY": WEEKLY, "DAILY": DAILY}
         return frequency_map[self.name]
 
@@ -52,19 +69,44 @@ def as_dateutil(self) -> weekday:
 
 
 class Pattern(pydantic.BaseModel):
+    model_config = pydantic.ConfigDict(validate_assignment=True)
+
     value: int
     type: PatternType
 
+    def __str__(self) -> str:
+        if self.value == -1:
+            qualifier = "last"
+        else:
+            qualifier = day_to_ordinal(self.value)
+        type_str = ""
+        if self.type != PatternType.DAY:
+            type_str = f" {self.type.name.lower().capitalize()}"
+        elif self.value == -1:
+            type_str = " day"
+        return f"{qualifier}{type_str}"
+
 
 class Schedule(pydantic.BaseModel):
+    """
+    Implements basic schedules. They are described in https://actualbudget.org/docs/budgeting/schedules/
+
+    Schedules are part of a rule which then compares if the date found would fit within the schedule by the .is_approx()
+    method. If it does, and the other conditions match, the transaction will then be linked with the schedule id
+    (stored in the database).
+    """
+
+    model_config = pydantic.ConfigDict(validate_assignment=True)
+
     start: datetime.date = pydantic.Field(..., description="Start date of the schedule.")
     interval: int = pydantic.Field(1, description="Repeat every interval at frequency unit.")
     frequency: Frequency = pydantic.Field(Frequency.MONTHLY, description="Unit for the defined interval.")
     patterns: list[Pattern] = pydantic.Field(default_factory=list)
     skip_weekend: bool = pydantic.Field(
-        alias="skipWeekend", description="If should move schedule before or after a weekend."
+        False, alias="skipWeekend", description="If should move schedule before or after a weekend."
     )
     weekend_solve_mode: WeekendSolveMode = pydantic.Field(
+        WeekendSolveMode.AFTER,
         alias="weekendSolveMode",
         description="When skipping weekend, the value should be set before or after the weekend interval.",
     )
@@ -74,14 +116,65 @@ class Schedule(pydantic.BaseModel):
         description="If the schedule should run forever or end at a certain date or number of occurrences.",
     )
     end_occurrences: int = pydantic.Field(
-        WeekendSolveMode.AFTER, alias="endOccurrences", description="Number of occurrences before the schedule ends."
+        1, alias="endOccurrences", description="Number of occurrences before the schedule ends."
     )
-    end_date: datetime.date = pydantic.Field(alias="endDate")
-
-    def is_approx(self, date: datetime.date) -> bool:
-        pass
+    end_date: datetime.date = pydantic.Field(None, alias="endDate")
+
+    def __str__(self) -> str:
+        # evaluate frequency: handle the case where DAILY convert to 'dai' instead of 'day'
+        interval = "day" if self.frequency == Frequency.DAILY else self.frequency.value.rstrip("ly")
+        frequency = interval if self.interval == 1 else f"{self.interval} {interval}s"
+        # evaluate
+        if self.frequency == Frequency.YEARLY:
+            target = f" on {self.start.strftime('%b %d')}"
+        elif self.frequency == Frequency.MONTHLY:
+            if not self.patterns:
+                target = f" on the {day_to_ordinal(self.start.day)}"
+            else:
+                patterns_str = []
+                for pattern in self.patterns:
+                    patterns_str.append(str(pattern))
+                target = " on the " + ", ".join(patterns_str)
+        elif self.frequency == Frequency.WEEKLY:
+            target = f" on {self.start.strftime('%A')}"
+        else:  # DAILY
+            target = ""
+        # end date part
+        if self.end_mode == EndMode.ON_DATE:
+            end = f", until {self.end_date}"
+        elif self.end_mode == EndMode.AFTER_N_OCCURRENCES:
+            end = ", once" if self.end_occurrences == 1 else f", {self.end_occurrences} times"
+        else:
+            end = ""
+        # weekend skips
+        move = f" ({self.weekend_solve_mode.value} weekend)" if self.skip_weekend else ""
+        return f"Every {frequency}{target}{end}{move}"
+
+    @pydantic.model_validator(mode="after")
+    def validate_end_date(self):
+        if self.end_mode == EndMode.ON_DATE and self.end_date is None:
+            raise ValueError("endDate cannot be 'None' when ")
+        if self.end_date is None:
+            self.end_date = self.start
+        return self
+
+    def is_approx(self, date: datetime.date, interval: datetime.timedelta = datetime.timedelta(days=2)) -> bool:
+        """This function checks if the input date could fit inside of this schedule. It will use the interval as the
+        maximum threshold before and after the specified date to look for. This defaults on Actual to 2 days."""
+        if date < self.start or (self.end_mode == EndMode.ON_DATE and self.end_date < date):
+            return False
+        before = self.before(date)
+        after = self.xafter(date, 1)
+        if before and (before - interval <= date <= before + interval):
+            return True
+        if after and (after[0] - interval <= date <= after[0] + interval):
+            return True
+        return False
 
     def rruleset(self) -> rruleset:
+        """Returns the rruleset from dateutil library. This is used internally to calculate the schedule dates.
+
+        For information on how to use this object check the official documentation https://dateutil.readthedocs.io"""
         rule_sets_configs = []
         config = dict(freq=self.frequency.as_dateutil(), dtstart=self.start, interval=self.interval)
         # add termination options
@@ -96,11 +189,14 @@ def rruleset(self) -> rruleset:
                     by_month_day.append(p.value)
                 else:  # it's a weekday
                     by_weekday.append(p.type.as_dateutil()(p.value))
+            # for the month or weekday rules, add a different rrule to the ruleset. This is because otherwise the rule
+            # would only look for, for example, days that are 15 that are also Fridays, and that is not desired
             if by_month_day:
                 monthly_config = config.copy() | {"bymonthday": by_month_day}
                 rule_sets_configs.append(monthly_config)
             if by_weekday:
                 rule_sets_configs.append(config.copy() | {"byweekday": by_weekday})
+        # if ruleset does not contain multiple rules, add the current rule as default
         if not rule_sets_configs:
             rule_sets_configs.append(config)
         # create rule set
@@ -109,6 +205,33 @@ def rruleset(self) -> rruleset:
             rs.rrule(rrule(**cfg))
         return rs
 
+    def do_skip_weekend(
+        self, dt_start: datetime.datetime, value: datetime.datetime
+    ) -> typing.Optional[datetime.datetime]:
+        if value.weekday() in (5, 6) and self.skip_weekend:
+            if self.weekend_solve_mode == WeekendSolveMode.AFTER:
+                value = value + datetime.timedelta(days=7 - value.weekday())
+                if self.end_mode == EndMode.ON_DATE and value > date_to_datetime(self.end_date):
+                    return None
+            else:  # BEFORE
+                value_after = value - datetime.timedelta(days=value.weekday() - 4)
+                if value_after < dt_start:
+                    # value is in the past, skip and look for another
+                    return None
+                value = value_after
+        return value
+
+    def before(self, date: datetime.date = None) -> typing.Optional[datetime.date]:
+        if not date:
+            date = datetime.date.today()
+        dt_start = date_to_datetime(date)
+        # we also always use the day before since today can also be a valid entry for our time
+        rs = self.rruleset()
+        before_datetime = rs.before(dt_start)
+        if not before_datetime:
+            return None
+        return self.do_skip_weekend(dt_start, before_datetime).date()
+
     def xafter(self, date: datetime.date = None, count: int = 1) -> list[datetime.date]:
         if not date:
             date = datetime.date.today()
@@ -117,22 +240,11 @@ def xafter(self, date: datetime.date = None, count: int = 1) -> list[datetime.da
         # we also always use the day before since today can also be a valid entry for our time
         rs = self.rruleset()
 
-        ret, i = [], 0
-        for value in rs:
-            value: datetime.datetime
-            if value.weekday() in (5, 6) and self.skip_weekend:
-                if self.weekend_solve_mode == WeekendSolveMode.AFTER:
-                    value = value + datetime.timedelta(days=7 - value.weekday())
-                else:  # BEFORE
-                    value_after = value - datetime.timedelta(days=value.weekday() - 4)
-                    if value_after < dt_start:
-                        # value is in the past, skip and look for another
-                        continue
-                    value = value_after
-            i += 1
-            dt = value
-            # convert back to date
-            ret.append(dt.date())
+        ret = []
+        for value in rs.xafter(dt_start, count, inc=True):
+            if value := self.do_skip_weekend(dt_start, value):
+                # convert back to date
+                ret.append(value.date())
             if len(ret) == count:
                 break
         return sorted(ret)

tests/test_schedules.py

@@ -1,9 +1,32 @@
-import datetime
+from datetime import date
 
-from actual.schedules import Schedule
+import pytest
+
+from actual.schedules import Schedule, date_to_datetime
 
 
 def test_basic_schedules():
+    s = Schedule.parse_obj(
+        {
+            "start": "2024-05-12",
+            "frequency": "monthly",
+            "skipWeekend": False,
+            "endMode": "after_n_occurrences",
+            "endOccurrences": 3,
+            "interval": 1,
+        }
+    )
+    assert s.before(date(2024, 5, 13)) == date(2024, 5, 12)
+    assert s.xafter(date(2024, 5, 12), 4) == [
+        date(2024, 5, 12),
+        date(2024, 6, 12),
+        date(2024, 7, 12),
+    ]
+
+    assert str(s) == "Every month on the 12th, 3 times"
+
+
+def test_complex_schedules():
     s = Schedule.parse_obj(
         {
             "start": "2024-05-08",
@@ -23,20 +46,92 @@ def test_basic_schedules():
             "interval": 1,
         }
     )
-    assert s.xafter(datetime.date(2024, 5, 10), count=5) == [
-        datetime.date(2024, 5, 10),
-        datetime.date(2024, 5, 13),
-        datetime.date(2024, 5, 27),
-        datetime.date(2024, 5, 31),
-        datetime.date(2024, 6, 5),
+    assert s.xafter(date(2024, 5, 10), count=5) == [
+        date(2024, 5, 10),
+        date(2024, 5, 13),
+        date(2024, 5, 27),
+        date(2024, 5, 31),
+        date(2024, 6, 5),
     ]
     # change the solve mode to before
     s.weekend_solve_mode = "before"
-    assert s.xafter(datetime.date(2024, 5, 10), count=5) == [
-        datetime.date(2024, 5, 10),
+    assert s.xafter(date(2024, 5, 10), count=5) == [
+        date(2024, 5, 10),
         # according to frontend, this entry happens twice
-        datetime.date(2024, 5, 10),
-        datetime.date(2024, 5, 24),
-        datetime.date(2024, 5, 31),
-        datetime.date(2024, 6, 5),
+        date(2024, 5, 10),
+        date(2024, 5, 24),
+        date(2024, 5, 31),
+        date(2024, 6, 5),
     ]
+
+    assert str(s) == "Every month on the last Sunday, 2nd Saturday, 10th, 31st, 5th (before weekend)"
+
+
+def test_is_approx():
+    # create schedule for every 1st and last day of the month (30th or 31st)
+    s = Schedule.parse_obj(
+        {
+            "start": "2024-05-10",
+            "frequency": "monthly",
+            "patterns": [
+                {"value": 1, "type": "day"},
+                {"value": -1, "type": "day"},
+            ],
+            "skipWeekend": True,
+            "weekendSolveMode": "after",
+            "endMode": "on_date",
+            "endOccurrences": 1,
+            "endDate": "2024-07-01",
+            "interval": 1,
+        }
+    )
+    # make sure the xafter is correct
+    assert s.xafter(date(2024, 6, 1), 5) == [
+        date(2024, 6, 3),
+        date(2024, 7, 1),
+        date(2024, 7, 1),
+    ]
+    # compare is_approx
+    assert s.is_approx(date(2024, 5, 1)) is False  # before starting period
+    assert s.is_approx(date(2024, 5, 30)) is True
+    assert s.is_approx(date(2024, 5, 31)) is True
+    assert s.is_approx(date(2024, 6, 1)) is True
+    assert s.is_approx(date(2024, 6, 3)) is True  # because 1st is also included
+
+    # 30th June is a sunday, so the right date would be 1st of June
+    assert s.is_approx(date(2024, 6, 28)) is False
+    assert s.is_approx(date(2024, 6, 30)) is True
+    assert s.is_approx(date(2024, 7, 1)) is True
+
+    # after end date we reject everything
+    assert s.is_approx(date(2024, 7, 2)) is False
+    assert s.is_approx(date(2024, 7, 31)) is False
+
+    assert str(s) == "Every month on the 1st, last day, until 2024-07-01 (after weekend)"
+
+
+def test_date_to_datetime():
+    dt = date(2024, 5, 1)
+    assert date_to_datetime(dt).date() == dt
+    assert date_to_datetime(None) is None
+
+
+def test_exceptions():
+    with pytest.raises(ValueError):
+        # on_date is set but no date is provided
+        Schedule.parse_obj(
+            {
+                "start": "2024-05-12",
+                "frequency": "monthly",
+                "skipWeekend": False,
+                "endMode": "on_date",
+                "endOccurrences": 3,
+                "interval": 1,
+            }
+        )
+
+
+def test_strings():
+    assert str(Schedule(start="2024-05-12", frequency="yearly")) == "Every year on May 12"
+    assert str(Schedule(start="2024-05-12", frequency="weekly")) == "Every week on Sunday"
+    assert str(Schedule(start="2024-05-12", frequency="daily")) == "Every day"