#!/usr/bin/env python3 """ Steward Profile - Tracks and learns user preferences and working patterns. This module builds a profile of the steward (user) to better understand their preferences, working style, and patterns. """ import json import os from datetime import datetime from typing import Dict, List, Any, Optional from collections import defaultdict from pathlib import Path import sys sys.path.insert(0, str(Path(__file__).parent.parent)) from config import MEMORY_DIR from utils.logging import setup_logger from intelligence.deep_behavioral_analysis import DeepBehavioralAnalyzer, create_behavioral_profile logger = setup_logger(__name__) class StewardProfile: """Manages and learns about the steward's preferences and patterns.""" def __init__(self): self.profile_file = os.path.join(MEMORY_DIR, "steward_profile.json") self.profile = self._load_profile() # Initialize deep behavioral analyzer self.behavioral_analyzer = DeepBehavioralAnalyzer() def _load_profile(self) -> Dict[str, Any]: """Load the steward profile from disk.""" if os.path.exists(self.profile_file): try: with open(self.profile_file, 'r') as f: return json.load(f) except Exception as e: logger.error(f"Error loading steward profile: {e}") # Default profile return { "name": "Unknown", "created_at": datetime.now().isoformat(), "communication_style": "direct", "technical_preferences": { "coding_style": "clean_and_documented", "testing_preference": "comprehensive", "documentation_style": "detailed" }, "work_patterns": { "peak_hours": [], "common_tasks": [], "project_types": [] }, "preferences": { "likes_examples": True, "detail_level": "balanced", "interaction_style": "collaborative" }, "common_patterns": [], "learning_history": [] } def _save_profile(self): """Save the steward profile to disk.""" try: os.makedirs(os.path.dirname(self.profile_file), exist_ok=True) with open(self.profile_file, 'w') as f: json.dump(self.profile, f, indent=2) except Exception as e: logger.error(f"Error saving steward profile: {e}") def update_from_message(self, message: str, context: Optional[Dict] = None): """Learn from a user message using deep behavioral analysis.""" try: # Perform deep behavioral analysis analysis = self.behavioral_analyzer.analyze_message(message, context) # Update profile with insights insights = analysis.get("immediate_insights", {}) patterns = analysis.get("patterns", {}) # Update communication style if insights.get("communication_style"): self.profile["communication_style"] = insights["communication_style"] # Update technical preferences from deep analysis if patterns.get("technical_preferences"): tech_prefs = patterns["technical_preferences"] if tech_prefs.get("preferred_languages"): self.profile["technical_preferences"]["preferred_languages"] = [ lang for lang, _ in tech_prefs["preferred_languages"] ] if tech_prefs.get("valued_practices"): self.profile["technical_preferences"]["valued_practices"] = [ practice for practice, _ in tech_prefs["valued_practices"] ] # Update work patterns if patterns.get("work_rhythm"): rhythm = patterns["work_rhythm"] if rhythm.get("peak_hours"): self.profile["work_patterns"]["peak_hours"] = rhythm["peak_hours"] if rhythm.get("work_style"): self.profile["work_patterns"]["style"] = rhythm["work_style"] # Update emotional patterns if insights.get("emotional_state"): if "emotional_patterns" not in self.profile: self.profile["emotional_patterns"] = {} self.profile["emotional_patterns"]["current_state"] = insights["emotional_state"] self.profile["emotional_patterns"]["stress_level"] = insights.get("urgency_level", "normal") # Add to learning history with deep analysis self.profile["learning_history"].append({ "timestamp": datetime.now().isoformat(), "message_snippet": message[:100], "context": context or {}, "behavioral_analysis": analysis }) # Keep only recent history self.profile["learning_history"] = self.profile["learning_history"][-100:] # Periodically create comprehensive profile if len(self.profile["learning_history"]) % 10 == 0: self._update_comprehensive_profile() self._save_profile() except Exception as e: logger.error(f"Error updating steward profile: {e}") def get_profile(self) -> Dict[str, Any]: """Get the current steward profile.""" return self.profile.copy() def get_preference(self, category: str, key: str, default: Any = None) -> Any: """Get a specific preference.""" return self.profile.get(category, {}).get(key, default) def update_preference(self, category: str, key: str, value: Any): """Update a specific preference.""" if category not in self.profile: self.profile[category] = {} self.profile[category][key] = value self._save_profile() def add_common_pattern(self, pattern: str): """Add a common pattern observed from the steward.""" if pattern not in self.profile["common_patterns"]: self.profile["common_patterns"].append(pattern) # Keep only top 20 patterns self.profile["common_patterns"] = self.profile["common_patterns"][-20:] self._save_profile() def _update_comprehensive_profile(self): """Update the profile with comprehensive behavioral analysis.""" try: # Create comprehensive behavioral profile behavioral_profile = create_behavioral_profile(self.behavioral_analyzer) if behavioral_profile.get("status") != "insufficient_data": # Update profile with comprehensive insights self.profile["behavioral_profile"] = behavioral_profile self.profile["last_comprehensive_update"] = datetime.now().isoformat() # Extract key insights for quick access if "personality_profile" in behavioral_profile: personality = behavioral_profile["personality_profile"] self.profile["communication_style"] = personality.get("communication_style", self.profile["communication_style"]) self.profile["work_patterns"]["style"] = personality.get("work_style", "flexible") self.profile["decision_style"] = personality.get("decision_style", "balanced") logger.info("Updated comprehensive behavioral profile") except Exception as e: logger.error(f"Error updating comprehensive profile: {e}") def get_behavioral_insights(self) -> Dict[str, Any]: """Get deep behavioral insights about the steward.""" return self.profile.get("behavioral_profile", {}) def get_current_emotional_state(self) -> Dict[str, str]: """Get current emotional state and stress level.""" return self.profile.get("emotional_patterns", { "current_state": "unknown", "stress_level": "normal" }) def get_work_recommendations(self) -> List[str]: """Get personalized work recommendations based on patterns.""" recommendations = [] # Based on work style work_style = self.profile.get("work_patterns", {}).get("style", "unknown") if work_style == "morning_person": recommendations.append("Schedule complex tasks for morning hours when you're most productive") elif work_style == "night_owl": recommendations.append("Reserve creative work for evening/night sessions") # Based on communication style comm_style = self.profile.get("communication_style", "direct") if comm_style == "collaborative": recommendations.append("Consider pair programming or team brainstorming sessions") elif comm_style == "questioning": recommendations.append("Deep dive documentation and exploratory tools might help satisfy curiosity") # Based on technical preferences practices = self.profile.get("technical_preferences", {}).get("valued_practices", []) if "testing" in practices: recommendations.append("TDD approach would align with your testing focus") if "documentation" in practices: recommendations.append("Consider documentation-first development approach") return recommendations