diff --git a/.gitignore b/.gitignore index 02eac69..2b346a0 100644 --- a/.gitignore +++ b/.gitignore @@ -1,22 +1,13 @@ -### AL ### -#Template for AL projects for Dynamics 365 Business Central -#launch.json folder +__pycache__/ +*.py[cod] +*$py.class +.pytest_cache/ +.coverage +htmlcov/ +.env +*.log +dist/ +build/ +*.egg-info/ .vscode/ -#Cache folder -.alcache/ -#Symbols folder -.alpackages/ -#Snapshots folder -.snapshots/ -#Testing Output folder -.output/ -#Extension App-file -*.app -#Rapid Application Development File -rad.json -#Translation Base-file -*.g.xlf -#License-file -*.flf -#Test results file -TestResults.xml \ No newline at end of file +.idea/ \ No newline at end of file diff --git a/src/iteration_state.py b/src/iteration_state.py new file mode 100644 index 0000000..5c57731 --- /dev/null +++ b/src/iteration_state.py @@ -0,0 +1,171 @@ +from __future__ import annotations +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum, auto +from typing import Any, Dict, Optional +import json +import os + + +class IterationStatus(Enum): + """ + Represents the possible states of a learning iteration. + """ + PENDING = auto() + RUNNING = auto() + COMPLETED = auto() + FAILED = auto() + INTERRUPTED = auto() + + +@dataclass +class IterationState: + """ + Represents the state of a single learning iteration. + + Provides comprehensive tracking of iteration details, including + configuration, performance metrics, status, and context. + """ + iteration_id: str + status: IterationStatus = IterationStatus.PENDING + start_time: Optional[datetime] = None + end_time: Optional[datetime] = None + configuration: Dict[str, Any] = field(default_factory=dict) + metrics: Dict[str, Any] = field(default_factory=dict) + error_details: Optional[str] = None + context: Dict[str, Any] = field(default_factory=dict) + + def mark_started(self) -> None: + """ + Mark the iteration as started and record the start time. + """ + self.status = IterationStatus.RUNNING + self.start_time = datetime.now() + + def mark_completed(self) -> None: + """ + Mark the iteration as completed and record the end time. + """ + self.status = IterationStatus.COMPLETED + self.end_time = datetime.now() + + def mark_failed(self, error_message: str) -> None: + """ + Mark the iteration as failed and store error details. + + Args: + error_message: Description of the failure + """ + self.status = IterationStatus.FAILED + self.end_time = datetime.now() + self.error_details = error_message + + def mark_interrupted(self) -> None: + """ + Mark the iteration as interrupted. + """ + self.status = IterationStatus.INTERRUPTED + self.end_time = datetime.now() + + def to_dict(self) -> Dict[str, Any]: + """ + Convert the iteration state to a dictionary representation. + + Returns: + Dict representation of the iteration state + """ + return { + 'iteration_id': self.iteration_id, + 'status': self.status.name, + 'start_time': self.start_time.isoformat() if self.start_time else None, + 'end_time': self.end_time.isoformat() if self.end_time else None, + 'configuration': self.configuration, + 'metrics': self.metrics, + 'error_details': self.error_details, + 'context': self.context + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> IterationState: + """ + Create an IterationState instance from a dictionary. + + Args: + data: Dictionary containing iteration state data + + Returns: + Reconstructed IterationState + """ + state = cls(iteration_id=data['iteration_id']) + state.status = IterationStatus[data['status']] + state.start_time = datetime.fromisoformat(data['start_time']) if data['start_time'] else None + state.end_time = datetime.fromisoformat(data['end_time']) if data['end_time'] else None + state.configuration = data['configuration'] + state.metrics = data['metrics'] + state.error_details = data['error_details'] + state.context = data['context'] + return state + + +class IterationStateManager: + """ + Manages the storage, retrieval, and tracking of iteration states. + """ + def __init__(self, state_dir: str = 'iteration_states'): + """ + Initialize the IterationStateManager. + + Args: + state_dir: Directory to store iteration states + """ + self.state_dir = state_dir + os.makedirs(state_dir, exist_ok=True) + + def save_state(self, iteration_state: IterationState) -> None: + """ + Save an iteration state to a JSON file. + + Args: + iteration_state: State to save + """ + state_file = os.path.join(self.state_dir, f'{iteration_state.iteration_id}.json') + with open(state_file, 'w') as f: + json.dump(iteration_state.to_dict(), f, indent=2) + + def load_state(self, iteration_id: str) -> Optional[IterationState]: + """ + Load an iteration state from a file. + + Args: + iteration_id: ID of the iteration to load + + Returns: + Loaded IterationState or None if not found + """ + state_file = os.path.join(self.state_dir, f'{iteration_id}.json') + try: + with open(state_file, 'r') as f: + state_data = json.load(f) + return IterationState.from_dict(state_data) + except FileNotFoundError: + return None + + def get_states_by_status(self, status: IterationStatus) -> List[IterationState]: + """ + Retrieve all iteration states with a specific status. + + Args: + status: Status to filter by + + Returns: + List of matching IterationState instances + """ + states = [] + for filename in os.listdir(self.state_dir): + if filename.endswith('.json'): + state_path = os.path.join(self.state_dir, filename) + with open(state_path, 'r') as f: + state_data = json.load(f) + if IterationStatus[state_data['status']] == status: + states.append(IterationState.from_dict(state_data)) + return states \ No newline at end of file diff --git a/tests/test_iteration_state.py b/tests/test_iteration_state.py new file mode 100644 index 0000000..91ac2e1 --- /dev/null +++ b/tests/test_iteration_state.py @@ -0,0 +1,103 @@ +import os +import pytest +from src.iteration_state import ( + IterationState, + IterationStatus, + IterationStateManager +) +from datetime import datetime, timedelta + + +def test_iteration_state_creation(): + """Test basic iteration state creation and attributes.""" + state = IterationState(iteration_id='test_001') + assert state.iteration_id == 'test_001' + assert state.status == IterationStatus.PENDING + + +def test_iteration_state_marking(): + """Test marking iteration states.""" + state = IterationState(iteration_id='test_002') + + # Test marking as started + state.mark_started() + assert state.status == IterationStatus.RUNNING + assert state.start_time is not None + + # Test marking as completed + state.mark_completed() + assert state.status == IterationStatus.COMPLETED + assert state.end_time is not None + + +def test_iteration_state_error_handling(): + """Test handling of failed and interrupted states.""" + state = IterationState(iteration_id='test_003') + + # Test marking as failed + state.mark_failed('Test error occurred') + assert state.status == IterationStatus.FAILED + assert state.error_details == 'Test error occurred' + + # Test marking as interrupted + state = IterationState(iteration_id='test_004') + state.mark_interrupted() + assert state.status == IterationStatus.INTERRUPTED + + +def test_iteration_state_serialization(): + """Test state serialization and deserialization.""" + original_state = IterationState(iteration_id='test_005') + original_state.configuration = {'learning_rate': 0.01} + original_state.metrics = {'accuracy': 0.95} + original_state.mark_started() + + # Convert to dict + state_dict = original_state.to_dict() + + # Recreate from dict + reconstructed_state = IterationState.from_dict(state_dict) + + assert reconstructed_state.iteration_id == original_state.iteration_id + assert reconstructed_state.status == original_state.status + assert reconstructed_state.configuration == original_state.configuration + assert reconstructed_state.metrics == original_state.metrics + + +def test_iteration_state_manager(): + """Test IterationStateManager functionality.""" + temp_dir = 'test_iteration_states' + os.makedirs(temp_dir, exist_ok=True) + + try: + manager = IterationStateManager(state_dir=temp_dir) + + # Create and save a state + state = IterationState(iteration_id='test_006') + state.mark_started() + state.metrics = {'loss': 0.1} + manager.save_state(state) + + # Load the state + loaded_state = manager.load_state('test_006') + assert loaded_state is not None + assert loaded_state.iteration_id == 'test_006' + assert loaded_state.metrics == {'loss': 0.1} + + # Get states by status + manager.save_state(IterationState(iteration_id='test_007')) + failed_state = IterationState(iteration_id='test_008') + failed_state.mark_failed('Test failure') + manager.save_state(failed_state) + + pending_states = manager.get_states_by_status(IterationStatus.PENDING) + failed_states = manager.get_states_by_status(IterationStatus.FAILED) + + assert len(pending_states) > 0 + assert len(failed_states) > 0 + + finally: + # Clean up test directory + for filename in os.listdir(temp_dir): + os.remove(os.path.join(temp_dir, filename)) + os.rmdir(temp_dir) \ No newline at end of file