HR1 example with PRedylogic

example.py

#!/usr/bin/env python3
import datetime
from ulid import ULID
import pendulum
from pydantic import BaseModel, Field, computed_field
from predylogic import Registry, all_of, any_of
from typing import Callable, Annotated

MonthNum = Annotated[int, Field(ge=1, le=12)]
YearNum = Annotated[int, Field(ge=2025)]
QuarterNum = Annotated[int, Field(ge=1, le=4)]
NonNegativeInt = Annotated[int, Field(ge=0)]
NonNegativeFloat = Annotated[float, Field(ge=0)]
MonthMinutes = Annotated[NonNegativeInt, Field(le=31*24*60)]



def generate_ulid(prefix: str = '') -> str:
    return prefix + str(ULID())

def id_factory(prefix: str = '') -> Callable[[], str]:
    return lambda: generate_ulid(prefix)



class FactBase(BaseModel):
    id: str = Field(default_factory=id_factory("fact:"))
    revoked_at: datetime.datetime | None = None


class CBVIncomeFact(FactBase):
    month: MonthNum
    year: YearNum
    work_minutes: MonthMinutes | None = None
    amount_cents: NonNegativeInt | None = None

    @computed_field
    @property
    def start_date(self: 'CBVIncomeFact') -> datetime.date:
        return datetime.date(self.year, self.month, 1)

    @computed_field
    @property
    def end_date(self: 'CBVIncomeFact') -> datetime.date:
        date_in_month = pendulum.date(self.year, self.month, 15)
        return datetime.date(self.year, self.month, date_in_month.end_of("month").day)


class BirthdateFact(FactBase):
    birthdate: datetime.date

    def age_on(self: 'BirthdateFact', date: datetime.date) -> int:
        return pendulum.date(date.year, date.month, date.day).diff(self.birthdate).in_years()


# class Beneficiary(BaseModel):
#     id: str = Field(default_factory=id_factory("person:"))
#     name: str
#     # These would be in a DB association eventually
#     determinations: list[Determination] = Field(default_factory=list)
#     cbv_facts: list[CBVIncomeFact] = Field(default_factory=list)


class DeterminationMonth(BaseModel):
    month: MonthNum
    year: YearNum
    # These would be pulled from a DB in a real system
    income_facts: list[CBVIncomeFact] = Field(default_factory=list)


class Determination(BaseModel):
    id: str = Field(default_factory=id_factory("det:"))
    compliant: bool | None = None
    months: list[DeterminationMonth]
    lookback_start: datetime.date
    lookback_end: datetime.date
    
    # This would be pulled from a DB in a real system
    birthdate_fact: BirthdateFact | None = None

    
# Create some records

cbv_202701 = CBVIncomeFact(year=2027, month=1, work_minutes=60*8, amount_cents=300*100)
cbv_202702 = CBVIncomeFact(year=2027, month=2, work_minutes=60*53, amount_cents=590*100)
cbv_202703 = CBVIncomeFact(year=2027, month=3, work_minutes=60*90, amount_cents=1023*100)

dm_202701 = DeterminationMonth(year=2027, month=1, income_facts=[cbv_202701])
dm_202702 = DeterminationMonth(year=2027, month=2, income_facts=[cbv_202702])
dm_202703 = DeterminationMonth(year=2027, month=3, income_facts=[cbv_202703])

bd_fact = BirthdateFact(birthdate=datetime.date(1962, 4, 23))

determination = Determination(lookback_start=datetime.date(2027, 1, 1),
                              lookback_end=datetime.date(2027, 6, 30),
                              months=[dm_202701, dm_202702, dm_202703],
                              birthdate_fact=bd_fact)

# Initialize registries
registry = Registry[Determination]("determination_rules")
month_registry = Registry[DeterminationMonth]("determination_month_rules")

# Define atomic predicates

@month_registry.rule_def()
def meets_hours_requirement(ctx: DeterminationMonth, threshold:int = 80) -> bool:
    return any(cbv.work_minutes is not None and cbv.work_minutes >= threshold * 60
          for cbv in ctx.income_facts)

@month_registry.rule_def()
def meets_income_requirement(ctx: DeterminationMonth, threshold:int = 580) -> bool:
    return any(cbv.amount_cents is not None and cbv.amount_cents >= threshold*100
          for cbv in ctx.income_facts)
        

# Define income policy
month_policy = any_of(
    [
        meets_hours_requirement(),
        meets_income_requirement()
    ]
)


@registry.rule_def()
def any_month_complies(ctx: Determination) -> bool:
    # FIXME: Would be neat if we could roll up month traces into bigger trace
    return any(month_policy(month) for month in ctx.months)

@registry.rule_def()
def exempt_under_19(ctx: Determination) -> bool:
    # Under 19 at youngest date in period (ie, lookback start)
    return ctx.birthdate_fact.age_on(ctx.lookback_start) < 19
    
@registry.rule_def()
def exempt_65_or_older(ctx: Determination) -> bool:
    # Over 65 at oldest date in period (ie, lookback end)
    return ctx.birthdate_fact.age_on(ctx.lookback_end) >= 65

policy = any_of(
    [
        
        exempt_under_19(),
        exempt_65_or_older(),
        any_month_complies()
    ]
)

trace = ce_policy(determination, trace=True, short_circuit=False)
print(trace)