Unified Iteration Tracking and State Management Infrastructure

.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
+<<<<<<< HEAD
+.idea/
+*.log
+=======
+.idea/
+>>>>>>> pr-3-Santix1234-ALP-Looping

src/alp/__init__.py

@@ -0,0 +1,3 @@
+from .iteration_tracker import IterationTracker, IterationStatus
+
+__all__ = ['IterationTracker', 'IterationStatus']

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))

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