diff --git a/.gitignore b/.gitignore index 02eac69..1010c5b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,22 +1,13 @@ -### AL ### -#Template for AL projects for Dynamics 365 Business Central -#launch.json folder +__pycache__/ +*.pyc +*.pyo +*.pyd +.pytest_cache/ +.coverage +htmlcov/ +.env +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/alp_loop.py b/src/alp_loop.py new file mode 100644 index 0000000..1f10af1 --- /dev/null +++ b/src/alp_loop.py @@ -0,0 +1,166 @@ +from abc import ABC, abstractmethod +from typing import Any, Dict, Optional +import logging +from dataclasses import dataclass, field +from enum import Enum, auto + +class LoopStatus(Enum): + """Enum representing the status of the learning loop.""" + INITIALIZED = auto() + RUNNING = auto() + PAUSED = auto() + COMPLETED = auto() + FAILED = auto() + +@dataclass +class LoopMetrics: + """Dataclass to track loop performance and metrics.""" + iterations: int = 0 + total_runtime: float = 0.0 + current_performance: float = 0.0 + additional_metrics: Dict[str, Any] = field(default_factory=dict) + +class AdaptiveLearningProcessLoop(ABC): + """ + Abstract Base Class for Adaptive Learning Process Loop Mechanism. + + This class defines the core structure and interface for implementing + iterative, self-improving learning cycles with robust error handling + and performance tracking. + + Key Responsibilities: + - Define the core learning loop structure + - Provide hooks for configuration and initialization + - Manage loop status and lifecycle + - Track and report performance metrics + - Support error handling and logging + """ + + def __init__( + self, + max_iterations: Optional[int] = None, + logger: Optional[logging.Logger] = None + ): + """ + Initialize the Adaptive Learning Process Loop. + + Args: + max_iterations (Optional[int]): Maximum number of iterations allowed. + logger (Optional[logging.Logger]): Custom logger for tracking events. + """ + self._max_iterations = max_iterations or float('inf') + self._logger = logger or logging.getLogger(self.__class__.__name__) + + # Core loop state tracking + self._status = LoopStatus.INITIALIZED + self._metrics = LoopMetrics() + + # Initialize logging + self._configure_logging() + + def _configure_logging(self): + """Configure logging with standard formatting.""" + handler = logging.StreamHandler() + formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + handler.setFormatter(formatter) + self._logger.addHandler(handler) + self._logger.setLevel(logging.INFO) + + @abstractmethod + def _initialize(self) -> None: + """ + Abstract method to perform initialization before the loop starts. + + Implementations must set up any required resources, load configurations, + and prepare the learning environment. + + Raises: + RuntimeError: If initialization fails + """ + pass + + @abstractmethod + def _iteration(self) -> bool: + """ + Abstract method representing a single learning iteration. + + Performs one complete learning cycle and returns whether + the loop should continue. + + Returns: + bool: True if the loop should continue, False to terminate + + Raises: + Exception: For any critical errors during iteration + """ + pass + + def run(self) -> LoopMetrics: + """ + Execute the main learning loop with robust error handling. + + Returns: + LoopMetrics: Performance metrics from the completed loop + + Raises: + RuntimeError: If loop encounters unrecoverable errors + """ + try: + # Initialize loop + self._status = LoopStatus.RUNNING + self._initialize() + + # Main learning loop + should_continue = True + while ( + self._metrics.iterations < self._max_iterations and + should_continue + ): + should_continue = self._iteration() + + # Update metrics + self._metrics.iterations += 1 + + # Check termination conditions + if not should_continue: + break + + # Mark loop completion + self._status = ( + LoopStatus.COMPLETED + if should_continue + else LoopStatus.FAILED + ) + + return self._metrics + + except Exception as e: + self._status = LoopStatus.FAILED + self._logger.error(f"Learning loop failed: {e}") + raise RuntimeError(f"Unrecoverable error in learning loop: {e}") from e + + def pause(self): + """ + Pause the current learning loop if supported. + + Implementations may override for specific pause behavior. + """ + if self._status == LoopStatus.RUNNING: + self._status = LoopStatus.PAUSED + self._logger.info("Learning loop paused") + elif self._status == LoopStatus.INITIALIZED: + # Allow pause from initialized state + self._status = LoopStatus.PAUSED + self._logger.info("Learning loop paused before running") + + def resume(self): + """ + Resume a paused learning loop. + + Implementations may override for specific resume behavior. + """ + if self._status == LoopStatus.PAUSED: + self._status = LoopStatus.RUNNING + self._logger.info("Learning loop resumed") \ No newline at end of file diff --git a/tests/test_alp_loop.py b/tests/test_alp_loop.py new file mode 100644 index 0000000..a03339a --- /dev/null +++ b/tests/test_alp_loop.py @@ -0,0 +1,49 @@ +import pytest +import logging +from src.alp_loop import AdaptiveLearningProcessLoop, LoopStatus, LoopMetrics + +class ConcreteALPLoop(AdaptiveLearningProcessLoop): + def __init__(self, max_iterations=5): + super().__init__(max_iterations) + self.initialization_called = False + self.iteration_count = 0 + + def _initialize(self): + self.initialization_called = True + + def _iteration(self): + self.iteration_count += 1 + return self.iteration_count < self._max_iterations + +def test_alp_loop_initialization(): + loop = ConcreteALPLoop() + assert loop._status == LoopStatus.INITIALIZED + assert isinstance(loop._logger, logging.Logger) + +def test_alp_loop_run(): + loop = ConcreteALPLoop() + metrics = loop.run() + + assert loop._status == LoopStatus.FAILED + assert metrics.iterations == 5 # Full number of iterations + assert loop.initialization_called is True + assert isinstance(metrics, LoopMetrics) + +def test_alp_loop_pause_resume(): + loop = ConcreteALPLoop(max_iterations=10) + assert loop._status == LoopStatus.INITIALIZED + + loop.pause() + assert loop._status == LoopStatus.PAUSED + +def test_alp_loop_max_iterations(): + loop = ConcreteALPLoop(max_iterations=3) + metrics = loop.run() + + assert metrics.iterations == 3 # Full number of iterations + assert loop._status == LoopStatus.FAILED + +def test_alp_loop_logging(): + loop = ConcreteALPLoop() + assert isinstance(loop._logger, logging.Logger) + assert loop._logger.level == logging.INFO \ No newline at end of file