""" Utility functions for time parsing, conflict detection, and metrics computation. """ from datetime import datetime, timedelta from typing import List, Dict, Tuple, Optional # ─── Time Utilities ─────────────────────────────────────────────────────────── TIME_SLOTS = [ "08:00", "08:30", "09:00", "09:30", "10:00", "10:30", "11:00", "11:30", "12:00", "12:30", "13:00", "13:30", "14:00", "14:30", "15:00", "15:30", "16:00", "16:30", "17:00", "17:30", "18:00" ] DURATIONS = [30, 60, 90, 120] # minutes def parse_time(time_str: str) -> datetime: """Parse a HH:MM time string into a datetime object (date fixed to today).""" return datetime.strptime(time_str, "%H:%M") def time_to_minutes(time_str: str) -> int: """Convert HH:MM to total minutes since midnight.""" dt = parse_time(time_str) return dt.hour * 60 + dt.minute def minutes_to_time(minutes: int) -> str: """Convert total minutes since midnight to HH:MM string.""" h = minutes // 60 m = minutes % 60 return f"{h:02d}:{m:02d}" def get_end_time(start_time: str, duration_minutes: int) -> str: """Calculate end time given start time and duration in minutes.""" start_mins = time_to_minutes(start_time) end_mins = start_mins + duration_minutes return minutes_to_time(end_mins) def time_ranges_overlap( start1: str, dur1: int, start2: str, dur2: int ) -> bool: """Check if two time ranges overlap. Args: start1: Start time of first range (HH:MM). dur1: Duration of first range in minutes. start2: Start time of second range (HH:MM). dur2: Duration of second range in minutes. Returns: True if the ranges overlap. """ s1 = time_to_minutes(start1) e1 = s1 + dur1 s2 = time_to_minutes(start2) e2 = s2 + dur2 return s1 < e2 and s2 < e1 def advance_time_slot(current_time: str, steps: int = 1) -> str: """Advance to the next time slot(s).""" current_mins = time_to_minutes(current_time) new_mins = current_mins + (30 * steps) # Clamp to end of day new_mins = min(new_mins, time_to_minutes("18:00")) return minutes_to_time(new_mins) # ─── Conflict Detection ────────────────────────────────────────────────────── def build_conflict_graph(tasks: List[Dict]) -> Dict[int, List[int]]: """Build an adjacency list of task conflicts based on time overlaps. Args: tasks: List of task dicts with 'id', 'time', 'duration', 'status'. Returns: Dict mapping task_id → list of conflicting task_ids. """ scheduled = [ t for t in tasks if t["status"] in ("pending", "scheduled", "completed") ] conflicts: Dict[int, List[int]] = {t["id"]: [] for t in scheduled} for i, t1 in enumerate(scheduled): for t2 in scheduled[i + 1:]: if time_ranges_overlap( t1["time"], t1.get("duration", 30), t2["time"], t2.get("duration", 30) ): conflicts[t1["id"]].append(t2["id"]) conflicts[t2["id"]].append(t1["id"]) return conflicts def count_conflicts(tasks: List[Dict]) -> int: """Count the total number of unique conflict pairs.""" graph = build_conflict_graph(tasks) count = 0 seen = set() for tid, neighbors in graph.items(): for nid in neighbors: pair = (min(tid, nid), max(tid, nid)) if pair not in seen: seen.add(pair) count += 1 return count # ─── Metrics Computation ───────────────────────────────────────────────────── def compute_metrics(state_dict: Dict) -> Dict[str, float]: """Compute evaluation metrics from a terminal state. Metrics: - task_completion_rate: fraction of tasks completed. - high_priority_completion: fraction of high-priority tasks completed. - conflict_count: number of scheduling conflicts remaining. - message_response_rate: fraction of inbox messages replied to. - efficiency_score: weighted composite score (0–100). """ tasks = state_dict.get("tasks", []) inbox = state_dict.get("inbox", []) # Task completion total_tasks = len(tasks) if tasks else 1 completed = sum(1 for t in tasks if t["status"] == "completed") task_completion_rate = completed / total_tasks # High-priority completion high_tasks = [t for t in tasks if t["priority"] == "high"] high_completed = sum(1 for t in high_tasks if t["status"] == "completed") high_priority_completion = ( high_completed / len(high_tasks) if high_tasks else 1.0 ) # Conflicts conflict_count = count_conflicts(tasks) # Message response total_messages = len(inbox) if inbox else 1 replied = sum(1 for m in inbox if m.get("replied", False)) message_response_rate = replied / total_messages # Efficiency score (weighted composite) efficiency_score = ( task_completion_rate * 35 + high_priority_completion * 30 + message_response_rate * 20 + max(0, (1 - conflict_count / max(total_tasks, 1))) * 15 ) return { "task_completion_rate": round(task_completion_rate, 3), "high_priority_completion": round(high_priority_completion, 3), "conflict_count": conflict_count, "message_response_rate": round(message_response_rate, 3), "efficiency_score": round(efficiency_score, 1), } # ─── Task/Message Titles ───────────────────────────────────────────────────── MEETING_TITLES = [ "Q4 Strategy Review", "1:1 with Manager", "Sprint Planning", "Client Call — Acme Corp", "Board Presentation Prep", "Design Review", "Weekly Standup", "Investor Update", "Product Roadmap Sync", "Cross-team Alignment", "Budget Review Meeting", "Hiring Panel Interview", "Vendor Negotiation", "Architecture Review", "Marketing Campaign Kickoff" ] WORK_TITLES = [ "Finalize Q3 Report", "Review PR #247", "Update Documentation", "Prepare Slide Deck", "Analyze Sales Data", "Write Technical Spec", "Code Review Session", "Database Migration Plan", "Security Audit Follow-up", "Performance Optimization", "Deploy Hotfix v2.3.1", "Update CI/CD Pipeline", "Refactor Auth Module", "Write Unit Tests", "API Integration Testing" ] PERSONAL_TITLES = [ "Dentist Appointment", "Gym Session", "Lunch with Friend", "Pick Up Dry Cleaning", "Car Service Appointment", "Call Insurance Company", "Grocery Shopping", "Yoga Class", "Parent-Teacher Conference", "Home Repair — Plumber" ] MESSAGE_CONTENTS_URGENT = [ "Need the quarterly figures ASAP for the board meeting.", "Critical bug in production — customer-facing. Please advise.", "Client escalation: contract renewal at risk. Call me.", "Server outage alert: all hands on deck.", "Investor meeting moved to tomorrow. Need deck by EOD.", "Legal flagged compliance issue. Immediate review needed.", "VP requesting project status update within the hour.", ] MESSAGE_CONTENTS_NORMAL = [ "FYI: Updated the shared drive with new templates.", "Team lunch this Friday — please RSVP.", "Quick question about the API changes in v2.4.", "Sharing meeting notes from yesterday's sync.", "Reminder: timesheets due by Friday.", "New onboarding docs are ready for review.", "Coffee chat next week? Let me know your availability.", ] SENDERS = [ "CEO", "VP Engineering", "Product Manager", "HR Director", "CFO", "Team Lead", "Client — Acme Corp", "External Counsel", "Marketing Director", "CTO", "Board Member", "Direct Report — Alex", "Direct Report — Priya", "Colleague — Jordan", "Colleague — Sam" ]