diff --git a/.gitignore b/.gitignore index 02eac69..efb5462 100644 --- a/.gitignore +++ b/.gitignore @@ -1,22 +1,27 @@ -### AL ### -#Template for AL projects for Dynamics 365 Business Central -#launch.json folder +__pycache__/ +<<<<<<< HEAD +*.pyc +*.pyo +*.pyd +======= +*.py[cod] +*$py.class +>>>>>>> pr-3-Santix1234-ALP-Looping +.pytest_cache/ +.coverage +htmlcov/ +.env +<<<<<<< HEAD +======= +*.log +>>>>>>> pr-3-Santix1234-ALP-Looping +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 +<<<<<<< HEAD +.idea/ +*.log +======= +.idea/ +>>>>>>> pr-3-Santix1234-ALP-Looping diff --git a/src/alp/__init__.py b/src/alp/__init__.py new file mode 100644 index 0000000..b41df07 --- /dev/null +++ b/src/alp/__init__.py @@ -0,0 +1,3 @@ +from .iteration_tracker import IterationTracker, IterationStatus + +__all__ = ['IterationTracker', 'IterationStatus'] \ No newline at end of file diff --git a/src/alp/iteration_tracker.py b/src/alp/iteration_tracker.py new file mode 100644 index 0000000..ca30a97 --- /dev/null +++ b/src/alp/iteration_tracker.py @@ -0,0 +1,155 @@ +from typing import Dict, Any, Optional +from enum import Enum, auto +import logging +from src.iteration_state import IterationStatus + +class IterationTracker: + """ + A comprehensive iteration tracking mechanism for the Adaptive Learning Process. + + Manages iteration state, progression, and provides detailed tracking capabilities. + + Attributes: + _current_iteration (int): Current iteration number + _max_iterations (Optional[int]): Maximum number of iterations allowed + _status (IterationStatus): Current status of the iteration process + _metadata (Dict[str, Any]): Additional metadata for tracking + """ + + def __init__(self, max_iterations: Optional[int] = None): + """ + Initialize the IterationTracker. + + Args: + max_iterations (Optional[int], optional): Maximum number of iterations allowed. + Defaults to None (unlimited iterations). + """ + self._current_iteration = 0 + self._max_iterations = max_iterations + self._status = IterationStatus.INITIALIZED + self._metadata: Dict[str, Any] = {} + self._logger = logging.getLogger(self.__class__.__name__) + + def start(self) -> None: + """ + Start the iteration process, setting the initial state. + + Raises: + RuntimeError: If iterations have already been started. + """ + if self._status != IterationStatus.INITIALIZED: + raise RuntimeError("Iteration process has already been started.") + + self._status = IterationStatus.IN_PROGRESS + self._logger.info("Iteration process started.") + + def next_iteration(self) -> bool: + """ + Advance to the next iteration. + + Returns: + bool: True if iterations can continue, False otherwise. + + Raises: + RuntimeError: If iteration process is not in progress. + """ + if self._status != IterationStatus.IN_PROGRESS: + raise RuntimeError("Iteration process is not in progress.") + + self._current_iteration += 1 + + # Check if we've reached the max iterations + if (self._max_iterations is not None and + self._current_iteration > self._max_iterations): + # Completed, but decrement to show we're at max + self._current_iteration -= 1 + self.complete() + return False + + self._logger.info(f"Starting iteration {self._current_iteration}") + + # If we've reached the max iterations, prepare to complete on next call + if (self._max_iterations is not None and + self._current_iteration == self._max_iterations): + return True + + return True + + def complete(self) -> None: + """Mark the iteration process as completed.""" + self._status = IterationStatus.COMPLETED + self._logger.info("Iteration process completed successfully.") + + def terminate(self, reason: Optional[str] = None) -> None: + """ + Terminate the iteration process prematurely. + + Args: + reason (Optional[str], optional): Reason for termination. + """ + self._status = IterationStatus.TERMINATED + self._logger.warning(f"Iteration process terminated. Reason: {reason}") + + def error(self, error_details: Optional[str] = None) -> None: + """ + Mark the iteration process as encountering an error. + + Args: + error_details (Optional[str], optional): Details of the error. + """ + self._status = IterationStatus.ERROR + self._logger.error(f"Iteration process encountered an error: {error_details}") + + @property + def current_iteration(self) -> int: + """ + Get the current iteration number. + + Returns: + int: Current iteration number. + """ + return self._current_iteration + + @property + def status(self) -> IterationStatus: + """ + Get the current iteration status. + + Returns: + IterationStatus: Current status of the iteration process. + """ + return self._status + + def add_metadata(self, key: str, value: Any) -> None: + """ + Add metadata to the iteration tracking. + + Args: + key (str): Metadata key + value (Any): Metadata value + """ + self._metadata[key] = value + + def get_metadata(self, key: str, default: Optional[Any] = None) -> Optional[Any]: + """ + Retrieve metadata by key. + + Args: + key (str): Metadata key + default (Optional[Any], optional): Default value if key is not found + + Returns: + Optional[Any]: Metadata value or default + """ + return self._metadata.get(key, default) + + def is_iteration_allowed(self) -> bool: + """ + Check if another iteration is allowed. + + Returns: + bool: True if iteration can continue, False otherwise. + """ + return (self._status == IterationStatus.IN_PROGRESS and + (self._max_iterations is None or + self._current_iteration < self._max_iterations)) \ No newline at end of file diff --git a/src/iteration_state.py b/src/iteration_state.py new file mode 100644 index 0000000..a93feda --- /dev/null +++ b/src/iteration_state.py @@ -0,0 +1,183 @@ +from __future__ import annotations +from dataclasses import dataclass, field +from datetime import datetime +from enum import Enum, auto +from typing import Any, Dict, Optional, List +import json +import os +import logging + +class IterationStatus(Enum): + """ + Comprehensive enumeration of possible iteration statuses. + Combines statuses from both IterationState and IterationTracker. + """ + INITIALIZED = auto() # From tracker + PENDING = auto() # From state + IN_PROGRESS = auto() # From tracker + RUNNING = auto() # From state + COMPLETED = auto() # Common + FAILED = auto() # From state + ERROR = auto() # From tracker + INTERRUPTED = auto() # From state + TERMINATED = auto() # From tracker + +@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) + logger: logging.Logger = field(default_factory=lambda: logging.getLogger(__name__)) + + def mark_started(self) -> None: + """ + Mark the iteration as started and record the start time. + """ + self.status = IterationStatus.RUNNING + self.start_time = datetime.now() + self.logger.info(f"Iteration {self.iteration_id} started.") + + def mark_completed(self) -> None: + """ + Mark the iteration as completed and record the end time. + """ + self.status = IterationStatus.COMPLETED + self.end_time = datetime.now() + self.logger.info(f"Iteration {self.iteration_id} completed successfully.") + + 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 + self.logger.error(f"Iteration {self.iteration_id} failed: {error_message}") + + def mark_interrupted(self) -> None: + """ + Mark the iteration as interrupted. + """ + self.status = IterationStatus.INTERRUPTED + self.end_time = datetime.now() + self.logger.warning(f"Iteration {self.iteration_id} was interrupted.") + + 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 + self.logger = logging.getLogger(self.__class__.__name__) + 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) + self.logger.info(f"Saved iteration state for {iteration_state.iteration_id}") + + 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: + self.logger.warning(f"No state found for iteration {iteration_id}") + 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 diff --git a/tests/test_iteration_tracker.py b/tests/test_iteration_tracker.py new file mode 100644 index 0000000..68d4511 --- /dev/null +++ b/tests/test_iteration_tracker.py @@ -0,0 +1,91 @@ +import pytest +from src.alp.iteration_tracker import IterationTracker, IterationStatus + +def test_iteration_tracker_initialization(): + """Test basic initialization of IterationTracker.""" + tracker = IterationTracker() + assert tracker.current_iteration == 0 + assert tracker.status == IterationStatus.INITIALIZED + +def test_iteration_tracker_start(): + """Test starting the iteration process.""" + tracker = IterationTracker() + tracker.start() + assert tracker.status == IterationStatus.IN_PROGRESS + +def test_iteration_tracker_next_iteration(): + """Test advancing to next iteration.""" + tracker = IterationTracker() + tracker.start() + assert tracker.next_iteration() is True + assert tracker.current_iteration == 1 + +def test_iteration_tracker_max_iterations(): + """Test iteration limit.""" + tracker = IterationTracker(max_iterations=3) + tracker.start() + + # Simulate iterations + for _ in range(3): + assert tracker.next_iteration() is True + + # Fourth iteration should return False + assert tracker.next_iteration() is False + assert tracker.status == IterationStatus.COMPLETED + +def test_iteration_tracker_manual_complete(): + """Test manual completion of iterations.""" + tracker = IterationTracker() + tracker.start() + tracker.complete() + assert tracker.status == IterationStatus.COMPLETED + +def test_iteration_tracker_terminate(): + """Test termination of iterations.""" + tracker = IterationTracker() + tracker.start() + tracker.terminate(reason="Test termination") + assert tracker.status == IterationStatus.TERMINATED + +def test_iteration_tracker_error(): + """Test error handling in iterations.""" + tracker = IterationTracker() + tracker.start() + tracker.error(error_details="Test error") + assert tracker.status == IterationStatus.ERROR + +def test_iteration_tracker_metadata(): + """Test metadata management.""" + tracker = IterationTracker() + tracker.add_metadata("test_key", "test_value") + assert tracker.get_metadata("test_key") == "test_value" + assert tracker.get_metadata("nonexistent_key") is None + +def test_iteration_tracker_is_iteration_allowed(): + """Test iteration allowance checks.""" + tracker = IterationTracker(max_iterations=2) + assert not tracker.is_iteration_allowed() # Not started + + tracker.start() + assert tracker.is_iteration_allowed() + + tracker.next_iteration() + assert tracker.is_iteration_allowed() + + tracker.next_iteration() + assert not tracker.is_iteration_allowed() # Max iterations reached + +def test_iteration_tracker_invalid_state_errors(): + """Test error handling for invalid state transitions.""" + tracker = IterationTracker() + + # Trying to get next iteration before start + with pytest.raises(RuntimeError): + tracker.next_iteration() + + # Start the tracker + tracker.start() + + # Trying to start again + with pytest.raises(RuntimeError): + tracker.start() \ No newline at end of file