Implement Adaptive Learning Process State Management System

.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
+.idea/

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

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)