Switching custom workflow to prefect

pyproject.toml

@@ -83,6 +83,7 @@ icalendar = "^5.0.13"
 pyobjc-framework-contacts = "^10.3.1"
 pytz = "^2024.1"
 apscheduler = "^3.10.4"
+prefect = "^2.20.4"
 
 [tool.poetry.group.test.dependencies]  # https://python-poetry.org/docs/master/managing-dependencies/
 coverage = { extras = ["toml"], version = ">=7.2.5" }

src/dspygen/modules/reminder_motivation_module.py

@@ -0,0 +1,55 @@
+"""
+Module for generating motivational advice for completing daily reminders.
+"""
+import dspy
+from dspygen.utils.dspy_tools import init_dspy
+
+
+class GenerateReminderMotivation(dspy.Signature):
+    """
+    Generate motivational advice for completing daily reminders.
+    Creates concise, encouraging messages that are:
+    - Specific: Directly related to the tasks at hand.
+    - Measurable: Tied to task completion.
+    - Achievable: Realistic within the given timeframe.
+    - Relevant: Pertinent to the user's day.
+    - Time-bound: Clear deadlines with AM/PM times.
+
+    Example outputs:
+    - "Finish the report by 2 PM – you've tackled similar tasks quickly!"
+    - "Call the client at 3:30 PM – a quick win that moves things forward!"
+    - "Gym at 6 PM – a boost for the rest of your week!"
+    """
+    num_tasks = dspy.InputField(desc="Number of remaining tasks for the day.")
+    task_list = dspy.InputField(desc="List of task titles and due times in JSON format, e.g., "
+                                     "[{'title': 'Complete project report', 'due_time': '2 PM'}, ...].")
+    motivation = dspy.OutputField(desc="Compact motivational message to help with task completion. "
+                                       "Focuses on reducing cognitive load while providing SMART tips.")
+
+
+class ReminderMotivationModule(dspy.Module):
+    """ReminderMotivationModule"""
+
+    def __init__(self, **forward_args):
+        super().__init__()
+        self.forward_args = forward_args
+        self.output = None
+
+    def forward(self, num_tasks, task_list):
+        pred = dspy.Predict(GenerateReminderMotivation)
+        self.output = pred(num_tasks=num_tasks, task_list=task_list).motivation
+        return self.output
+
+def reminder_motivation_call(num_tasks: int, task_list: str):
+    motivation = ReminderMotivationModule()
+    return motivation.forward(num_tasks=num_tasks, task_list=task_list)
+
+def main():
+    init_dspy()
+    num_tasks = 3
+    task_list = "1. Complete project report (Due: 14:00)\n2. Call client (Due: 15:30)\n3. Gym session (Due: 18:00)"
+    result = reminder_motivation_call(num_tasks=num_tasks, task_list=task_list)
+    print(result)
+
+if __name__ == "__main__":
+    main()

src/dspygen/pyautomator/contacts/contacts_app.py

@@ -0,0 +1,150 @@
+import objc
+from Contacts import CNContactStore, CNContact, CNSaveRequest, CNContactFormatter
+from Foundation import NSPredicate
+from typing import List, Dict, Any, Optional
+
+from dspygen.pyautomator.base_app import BaseApp
+from dspygen.pyautomator.contacts.contacts_main import Contact, ContactError
+
+class ContactsApp(BaseApp):
+    def __init__(self):
+        super().__init__("Contacts")
+        self.store = CNContactStore.alloc().init()
+
+    def request_access(self):
+        """Request access to contacts."""
+        def callback(granted, error):
+            if not granted:
+                raise PermissionError("Access to contacts denied.")
+
+        self.store.requestAccessForEntityType_completionHandler_(CNEntityTypeContacts, callback)
+
+    def get_all_contacts(self) -> List[Contact]:
+        """Retrieve all contacts."""
+        keys_to_fetch = [
+            CNContactGivenNameKey,
+            CNContactFamilyNameKey,
+            CNContactEmailAddressesKey,
+            CNContactPhoneNumbersKey,
+            CNContactPostalAddressesKey,
+            CNContactBirthdayKey,
+            CNContactImageDataKey
+        ]
+
+        try:
+            contacts = Contact.fetch_contacts(NSPredicate.predicateWithValue_(True), keys_to_fetch)
+            return [Contact.from_cn_contact(contact) for contact in contacts]
+        except ContactError as e:
+            print(f"Error fetching contacts: {e}")
+            return []
+
+    def search_contacts(self, query: str) -> List[Contact]:
+        """Search contacts based on a query string."""
+        keys_to_fetch = [
+            CNContactGivenNameKey,
+            CNContactFamilyNameKey,
+            CNContactEmailAddressesKey,
+            CNContactPhoneNumbersKey
+        ]
+
+        predicate = CNContact.predicateForContactsMatchingName_(query)
+
+        try:
+            contacts = Contact.fetch_contacts(predicate, keys_to_fetch)
+            return [Contact.from_cn_contact(contact) for contact in contacts]
+        except ContactError as e:
+            print(f"Error searching contacts: {e}")
+            return []
+
+    def add_contact(self, contact: Contact) -> bool:
+        """Add a new contact."""
+        try:
+            contact.save()
+            return True
+        except ContactError as e:
+            print(f"Error adding contact: {e}")
+            return False
+
+    def update_contact(self, contact: Contact) -> bool:
+        """Update an existing contact."""
+        try:
+            save_request = CNSaveRequest.alloc().init()
+            save_request.updateContact_(contact.cn_contact)
+            self.store.executeSaveRequest_error_(save_request, None)
+            return True
+        except objc.error as e:
+            print(f"Error updating contact: {e}")
+            return False
+
+    def delete_contact(self, contact: Contact) -> bool:
+        """Delete a contact."""
+        try:
+            save_request = CNSaveRequest.alloc().init()
+            save_request.deleteContact_(contact.cn_contact)
+            self.store.executeSaveRequest_error_(save_request, None)
+            return True
+        except objc.error as e:
+            print(f"Error deleting contact: {e}")
+            return False
+
+    def get_contact_by_name(self, given_name: str, family_name: str) -> Optional[Contact]:
+        """Retrieve a contact by given name and family name."""
+        keys_to_fetch = [
+            CNContactGivenNameKey,
+            CNContactFamilyNameKey,
+            CNContactEmailAddressesKey,
+            CNContactPhoneNumbersKey,
+            CNContactPostalAddressesKey,
+            CNContactBirthdayKey,
+            CNContactImageDataKey
+        ]
+
+        predicate = CNContact.predicateForContactsMatchingName_(f"{given_name} {family_name}")
+
+        try:
+            contacts = Contact.fetch_contacts(predicate, keys_to_fetch)
+            if contacts:
+                return Contact.from_cn_contact(contacts[0])
+            return None
+        except ContactError as e:
+            print(f"Error fetching contact: {e}")
+            return None
+
+def main():
+    app = ContactsApp()
+    app.request_access()
+
+    # Example usage
+    print("All Contacts:")
+    all_contacts = app.get_all_contacts()
+    for contact in all_contacts[:5]:  # Print first 5 contacts
+        print(contact)
+
+    print("\nSearching for 'John':")
+    search_results = app.search_contacts("John")
+    for contact in search_results:
+        print(contact)
+
+    # Add a new contact
+    new_contact = Contact.create(
+        given_name="Jane",
+        family_name="Doe",
+        email_addresses=[("home", "[email protected]")],
+        phone_numbers=[("iPhone", "(555) 987-6543")]
+    )
+    if app.add_contact(new_contact):
+        print("\nNew contact added successfully")
+
+    # Update the contact
+    jane = app.get_contact_by_name("Jane", "Doe")
+    if jane:
+        jane.phone_numbers = [("iPhone", "(555) 123-4567")]
+        if app.update_contact(jane):
+            print("Contact updated successfully")
+
+    # Delete the contact
+    if jane and app.delete_contact(jane):
+        print("Contact deleted successfully")
+
+if __name__ == "__main__":
+    main()