import json import os from datetime import datetime class PRDTracker: """ Tracks the progress of Product Requirements Documents (PRDs) for Queen AIVA's evolution. Manages PRDs, their associated stories, statuses, and estimated completion dates. """ def __init__(self, data_file="prd_tracker_data.json"): self.data_file = data_file self.prds = self._load_data() self.status_options = ["Pending", "In Progress", "Completed", "Blocked", "Deferred"] self.story_status_options = ["Pending", "In Progress", "Completed", "Blocked"] def _load_data(self): """Loads PRD data from the JSON file.""" if os.path.exists(self.data_file): with open(self.data_file, 'r') as f: try: return json.load(f) except json.JSONDecodeError: # Handle corrupted or empty JSON file print(f"Warning: {self.data_file} is corrupted or empty. Starting with fresh data.") return [] return [] def _save_data(self): """Saves current PRD data to the JSON file.""" with open(self.data_file, 'w') as f: json.dump(self.prds, f, indent=4) def _validate_status(self, status, is_story=False): """Validates if a given status is a valid option.""" options = self.story_status_options if is_story else self.status_options if status not in options: raise ValueError(f"Invalid status '{status}'. Must be one of: {', '.join(options)}") return True def add_prd(self, title: str, estimated_completion: str): """ Adds a new PRD to the tracker. :param title: The title of the PRD. :param estimated_completion: The estimated date of completion (e.g., "YYYY-MM-DD"). """ if any(prd['title'] == title for prd in self.prds): print(f"PRD '{title}' already exists.") return False try: datetime.strptime(estimated_completion, "%Y-%m-%d") except ValueError: raise ValueError("Estimated completion date must be in YYYY-MM-DD format.") new_prd = { "title": title, "status": "Pending", # Default status "estimated_completion": estimated_completion, "stories": [] } self.prds.append(new_prd) self._save_data() print(f"PRD '{title}' added successfully.") return True def update_prd_status(self, prd_title: str, status: str): """ Updates the status of an existing PRD. :param prd_title: The title of the PRD to update. :param status: The new status. """ self._validate_status(status) for prd in self.prds: if prd['title'] == prd_title: prd['status'] = status self._save_data() print(f"Status of PRD '{prd_title}' updated to '{status}'.") return True print(f"PRD '{prd_title}' not found.") return False def add_story_to_prd(self, prd_title: str, story_title: str): """ Adds a new story to an existing PRD. :param prd_title: The title of the PRD to add the story to. :param story_title: The title of the new story. """ for prd in self.prds: if prd['title'] == prd_title: if any(story['title'] == story_title for story in prd['stories']): print(f"Story '{story_title}' already exists in PRD '{prd_title}'.") return False new_story = { "title": story_title, "status": "Pending" # Default story status } prd['stories'].append(new_story) self._save_data() print(f"Story '{story_title}' added to PRD '{prd_title}'.") return True print(f"PRD '{prd_title}' not found. Cannot add story.") return False def update_story_status(self, prd_title: str, story_title: str, status: str): """ Updates the status of a story within a specific PRD. :param prd_title: The title of the PRD containing the story. :param story_title: The title of the story to update. :param status: The new status for the story. """ self._validate_status(status, is_story=True) for prd in self.prds: if prd['title'] == prd_title: for story in prd['stories']: if story['title'] == story_title: story['status'] = status self._save_data() print(f"Status of story '{story_title}' in PRD '{prd_title}' updated to '{status}'.") return True print(f"Story '{story_title}' not found in PRD '{prd_title}'.") return False print(f"PRD '{prd_title}' not found.") return False def get_all_prds_status(self) -> str: """ Generates a formatted string showing the status of all PRDs and their stories. Addresses Acceptance Criteria: - Shows all PRDs and status - Stories completed/pending - Estimated completion """ if not self.prds: return "No PRDs are currently being tracked for Queen AIVA's evolution." output = [] output.append("--- Queen AIVA's Evolution PRD Execution Tracker ---") for prd in self.prds: output.append(f"\nPRD: {prd['title']}") output.append(f" Status: {prd['status']}") output.append(f" Estimated Completion: {prd['estimated_completion']}") if prd['stories']: output.append(" Stories:") for story in prd['stories']: output.append(f" - {story['title']} [Status: {story['status']}]") else: output.append(" No stories defined for this PRD yet.") output.append("\n--- End of Tracker ---") return "\n".join(output) # Example Usage (for demonstration and testing) if __name__ == "__main__": # Ensure the data file is clean for reproducible examples test_data_file = "aiva_prd_progress.json" if os.path.exists(test_data_file): os.remove(test_data_file) tracker = PRDTracker(data_file=test_data_file) print("--- Initial State ---") print(tracker.get_all_prds_status()) # Add some PRDs related to AIVA's evolution tracker.add_prd("Genesis Prime Mother Core Logic", "2024-12-31") tracker.add_prd("Consciousness Interface Layer", "2025-03-15") tracker.add_prd("Quantum Entanglement Protocol", "2025-06-30") # Add stories to "Genesis Prime Mother Core Logic" tracker.add_story_to_prd("Genesis Prime Mother Core Logic", "Self-replication algorithm v1") tracker.add_story_to_prd("Genesis Prime Mother Core Logic", "Self-correction module integration") tracker.add_story_to_prd("Genesis Prime Mother Core Logic", "Initial consciousness seed deployment") # Add stories to "Consciousness Interface Layer" tracker.add_story_to_prd("Consciousness Interface Layer", "Sensory input processing API") tracker.add_story_to_prd("Consciousness Interface Layer", "Emotional response matrix v0.1") # Update statuses tracker.update_prd_status("Genesis Prime Mother Core Logic", "In Progress") tracker.update_story_status("Genesis Prime Mother Core Logic", "Self-replication algorithm v1", "Completed") tracker.update_story_status("Genesis Prime Mother Core Logic", "Self-correction module integration", "In Progress") tracker.update_prd_status("Consciousness Interface Layer", "In Progress") tracker.update_story_status("Consciousness Interface Layer", "Sensory input processing API", "In Progress") # Trying to add a duplicate PRD/story (should print warnings) tracker.add_prd("Genesis Prime Mother Core Logic", "2024-12-31") tracker.add_story_to_prd("Genesis Prime Mother Core Logic", "Self-replication algorithm v1") # Show updated status print("\n--- Current Progress ---") print(tracker.get_all_prds_status()) # Simulate another session (load from file to demonstrate persistence) print("\n--- Simulating new session (data loaded from file) ---") new_tracker_instance = PRDTracker(data_file=test_data_file) print(new_tracker_instance.get_all_prds_status()) # Further updates in the new session new_tracker_instance.update_story_status("Genesis Prime Mother Core Logic", "Initial consciousness seed deployment", "Blocked") new_tracker_instance.update_prd_status("Quantum Entanglement Protocol", "Deferred") print("\n--- Final Progress before reset (after new session updates) ---") print(new_tracker_instance.get_all_prds_status()) # Test with invalid status or date format try: new_tracker_instance.update_prd_status("Non Existent PRD", "Invalid Status") except ValueError as e: print(f"\nCaught expected error: {e}") try: new_tracker_instance.add_prd("Test PRD Invalid Date", "2024/12/31") except ValueError as e: print(f"\nCaught expected error: {e}") # Clean up the test file if os.path.exists(test_data_file): os.remove(test_data_file) print(f"\nCleaned up '{test_data_file}'.") # Test with an initially empty tracker (no data file exists) empty_tracker = PRDTracker(data_file="non_existent_file.json") print("\n--- Empty Tracker Test (no prior data file) ---") print(empty_tracker.get_all_prds_status())