diff --git a/.gitignore b/.gitignore index 02eac69..f48151f 100644 --- a/.gitignore +++ b/.gitignore @@ -1,22 +1,52 @@ -### AL ### -#Template for AL projects for Dynamics 365 Business Central -#launch.json folder +<<<<<<< HEAD +__pycache__/ +*.py[cod] +*$py.class +.pytest_cache/ +.coverage +htmlcov/ +*.log +.env +dist/ +build/ +*.egg-info/ +.venv/ +venv/ +.idea/ .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 +======= +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so + +# Virtual Environments +venv/ +env/ +.env/ + +# Distribution / packaging +dist/ +build/ +*.egg-info/ + +# Logs +*.log +logs/ + +# Testing +.pytest_cache/ +.coverage +htmlcov/ + +# IDEs and editors +.vscode/ +.idea/ +*.swp +*.swo + +# OS generated files +.DS_Store +Thumbs.db +>>>>>>> pr-2-hubbahubba11x-ALP-Looping diff --git a/src/iteration_logger.py b/src/iteration_logger.py new file mode 100644 index 0000000..3fcd644 --- /dev/null +++ b/src/iteration_logger.py @@ -0,0 +1,173 @@ +import logging +from typing import Dict, Any, Optional, List +from dataclasses import dataclass, asdict, field +from datetime import datetime +import json +import os + + +@dataclass +class IterationLogEntry: + """ + Structured log entry for a single ALP iteration. + + Attributes: + iteration_number (int): Unique identifier for the iteration + timestamp (datetime): Timestamp of the iteration + metadata (Dict[str, Any]): Additional metadata about the iteration + performance_metrics (Dict[str, float]): Performance metrics for the iteration + error_info (Optional[Dict[str, Any]]): Error information if an error occurred + status (str): Status of the iteration (success, failure, warning) + """ + iteration_number: int + timestamp: datetime = field(default_factory=datetime.now) + metadata: Dict[str, Any] = field(default_factory=dict) + performance_metrics: Dict[str, float] = field(default_factory=dict) + error_info: Optional[Dict[str, Any]] = None + status: str = 'pending' + + def to_dict(self) -> Dict[str, Any]: + """ + Convert log entry to a dictionary for JSON serialization. + + Returns: + Dict[str, Any]: Serializable dictionary representation of the log entry + """ + entry_dict = asdict(self) + entry_dict['timestamp'] = self.timestamp.isoformat() + return entry_dict + + +class IterationLogger: + """ + Utility class for managing and writing log entries for ALP iterations. + + Supports file-based and console logging with configurable log levels. + """ + def __init__( + self, + log_dir: str = 'logs', + log_file: str = 'iteration_log.json', + console_log_level: int = logging.INFO + ): + """ + Initialize the IterationLogger. + + Args: + log_dir (str): Directory to store log files + log_file (str): Name of the log file + console_log_level (int): Logging level for console output + """ + self.log_dir = log_dir + self.log_file = log_file + + # Create logs directory if it doesn't exist + os.makedirs(log_dir, exist_ok=True) + + # Full path for log file + self.log_path = os.path.join(log_dir, log_file) + + # Configure console logging + logging.basicConfig( + level=console_log_level, + format='%(asctime)s - %(levelname)s: %(message)s' + ) + self.logger = logging.getLogger(__name__) + + def log_iteration(self, entry: IterationLogEntry) -> None: + """ + Log an iteration entry to file and console. + + Args: + entry (IterationLogEntry): Iteration log entry to record + """ + try: + # Log to console + self._log_to_console(entry) + + # Append to JSON log file + self._append_to_log_file(entry) + except Exception as e: + self.logger.error(f"Error logging iteration: {e}") + + def _log_to_console(self, entry: IterationLogEntry) -> None: + """ + Log iteration details to console based on status. + + Args: + entry (IterationLogEntry): Iteration log entry + """ + log_method = { + 'success': self.logger.info, + 'failure': self.logger.error, + 'warning': self.logger.warning, + 'pending': self.logger.debug + }.get(entry.status, self.logger.info) + + log_method( + f"Iteration {entry.iteration_number} " + f"Status: {entry.status} " + f"Metrics: {entry.performance_metrics}" + ) + + def _append_to_log_file(self, entry: IterationLogEntry) -> None: + """ + Append iteration log entry to JSON log file. + + Args: + entry (IterationLogEntry): Iteration log entry + """ + try: + # Read existing log entries or initialize empty list + log_entries = [] + if os.path.exists(self.log_path): + with open(self.log_path, 'r') as f: + log_entries = json.load(f) + + # Append new entry + log_entries.append(entry.to_dict()) + + # Write updated log entries + with open(self.log_path, 'w') as f: + json.dump(log_entries, f, indent=2) + except Exception as e: + self.logger.error(f"Failed to write log entry: {e}") + + def get_log_entries(self) -> list: + """ + Retrieve all log entries from the log file. + + Returns: + list: List of log entries + """ + try: + if os.path.exists(self.log_path): + with open(self.log_path, 'r') as f: + return json.load(f) + return [] + except Exception as e: + self.logger.error(f"Error reading log entries: {e}") + return [] + + def filter_log_entries(self, status: Optional[str] = None, min_iteration: Optional[int] = None) -> List[Dict[str, Any]]: + """ + Filter log entries based on status and minimum iteration number. + + Args: + status (Optional[str]): Filter entries by status (e.g., 'success', 'failure') + min_iteration (Optional[int]): Minimum iteration number to include + + Returns: + List[Dict[str, Any]]: Filtered log entries + """ + log_entries = self.get_log_entries() + + filtered_entries = log_entries.copy() + + if status is not None: + filtered_entries = [entry for entry in filtered_entries if entry.get('status') == status] + + if min_iteration is not None: + filtered_entries = [entry for entry in filtered_entries if entry.get('iteration_number', 0) >= min_iteration] + + return filtered_entries \ No newline at end of file diff --git a/src/log_analysis/__init__.py b/src/log_analysis/__init__.py new file mode 100644 index 0000000..289c1c2 --- /dev/null +++ b/src/log_analysis/__init__.py @@ -0,0 +1 @@ +from .log_parser import LogParser \ No newline at end of file diff --git a/src/log_analysis/log_parser.py b/src/log_analysis/log_parser.py new file mode 100644 index 0000000..49233d1 --- /dev/null +++ b/src/log_analysis/log_parser.py @@ -0,0 +1,130 @@ +from typing import List, Dict, Any +import json +import os +from datetime import datetime + +class LogParser: + """ + A class responsible for parsing and analyzing iteration logs. + + This class provides methods to: + - Read log files + - Parse log entries + - Analyze performance metrics + - Generate summary reports + """ + + def __init__(self, log_directory: str = 'logs'): + """ + Initialize the LogParser with a specific log directory. + + Args: + log_directory (str): Directory containing log files. Defaults to 'logs'. + """ + self.log_directory = log_directory + + # Ensure log directory exists + os.makedirs(log_directory, exist_ok=True) + + def get_log_files(self) -> List[str]: + """ + Retrieve all log files in the specified directory. + + Returns: + List[str]: List of log file paths + """ + return [ + os.path.join(self.log_directory, f) + for f in os.listdir(self.log_directory) + if f.endswith('.json') + ] + + def parse_log_file(self, file_path: str) -> List[Dict[str, Any]]: + """ + Parse a single log file and return its contents. + + Args: + file_path (str): Path to the log file + + Returns: + List[Dict[str, Any]]: Parsed log entries + + Raises: + FileNotFoundError: If the log file doesn't exist + json.JSONDecodeError: If the log file is not valid JSON + """ + try: + with open(file_path, 'r') as log_file: + return json.load(log_file) + except FileNotFoundError: + raise FileNotFoundError(f"Log file not found: {file_path}") + except json.JSONDecodeError: + raise ValueError(f"Invalid JSON in log file: {file_path}") + + def analyze_performance(self, log_entries: List[Dict[str, Any]]) -> Dict[str, Any]: + """ + Analyze performance metrics from log entries. + + Args: + log_entries (List[Dict[str, Any]]): List of log entries to analyze + + Returns: + Dict[str, Any]: Performance summary metrics + """ + if not log_entries: + return {} + + # Extract performance-related metrics + performance_scores = [] + for entry in log_entries: + # Check for performance_score in both top-level and nested locations + score = entry.get('performance_score', 0) + if isinstance(score, (int, float)): + performance_scores.append(score) + + performance_metrics = { + 'total_iterations': len(log_entries), + 'start_time': log_entries[0].get('timestamp'), + 'end_time': log_entries[-1].get('timestamp'), + 'error_rate': sum(1 for entry in log_entries if entry.get('status') == 'error') / len(log_entries) + } + + # Calculate performance metrics only if scores exist + if performance_scores: + performance_metrics.update({ + 'performance_scores': performance_scores, + 'avg_performance': sum(performance_scores) / len(performance_scores), + 'max_performance': max(performance_scores), + 'min_performance': min(performance_scores) + }) + + return performance_metrics + + def generate_report(self) -> Dict[str, Any]: + """ + Generate a comprehensive report by analyzing all log files. + + Returns: + Dict[str, Any]: Comprehensive log analysis report + """ + log_files = self.get_log_files() + report = { + 'total_log_files': len(log_files), + 'file_analyses': [] + } + + for log_file in log_files: + try: + log_entries = self.parse_log_file(log_file) + file_report = { + 'file_name': os.path.basename(log_file), + 'performance': self.analyze_performance(log_entries) + } + report['file_analyses'].append(file_report) + except Exception as e: + report['file_analyses'].append({ + 'file_name': os.path.basename(log_file), + 'error': str(e) + }) + + return report \ No newline at end of file diff --git a/tests/log_analysis/test_log_parser.py b/tests/log_analysis/test_log_parser.py new file mode 100644 index 0000000..da41fdf --- /dev/null +++ b/tests/log_analysis/test_log_parser.py @@ -0,0 +1,110 @@ +import os +import json +import pytest +from typing import List, Dict, Any +from src.log_analysis.log_parser import LogParser + +@pytest.fixture +def sample_log_data() -> List[Dict[str, Any]]: + """Generate sample log data for testing.""" + return [ + { + 'timestamp': '2023-01-01T00:00:00', + 'iteration': 1, + 'performance_score': 0.75, + 'status': 'success' + }, + { + 'timestamp': '2023-01-01T00:01:00', + 'iteration': 2, + 'performance_score': 0.85, + 'status': 'success' + }, + { + 'timestamp': '2023-01-01T00:02:00', + 'iteration': 3, + 'performance_score': 0.5, + 'status': 'error' + } + ] + +@pytest.fixture +def log_directory(tmp_path, sample_log_data): + """Create a temporary log directory with sample log files.""" + log_dir = tmp_path / "logs" + log_dir.mkdir() + + # Create multiple log files + for i in range(3): + log_file = log_dir / f"log_{i}.json" + with open(log_file, 'w') as f: + json.dump(sample_log_data, f) + + return str(log_dir) + +def test_log_parser_initialization(log_directory): + """Test LogParser initialization.""" + parser = LogParser(log_directory) + assert parser.log_directory == log_directory + assert os.path.exists(log_directory) + +def test_get_log_files(log_directory): + """Test retrieving log files.""" + parser = LogParser(log_directory) + log_files = parser.get_log_files() + + assert len(log_files) == 3 + assert all(f.endswith('.json') for f in log_files) + +def test_parse_log_file(log_directory, sample_log_data): + """Test parsing a single log file.""" + parser = LogParser(log_directory) + log_files = parser.get_log_files() + parsed_logs = parser.parse_log_file(log_files[0]) + + assert parsed_logs == sample_log_data + assert len(parsed_logs) == 3 + +def test_analyze_performance(log_directory, sample_log_data): + """Test performance analysis of log entries.""" + parser = LogParser(log_directory) + performance_metrics = parser.analyze_performance(sample_log_data) + + assert performance_metrics['total_iterations'] == 3 + assert performance_metrics['error_rate'] == pytest.approx(1/3) + assert performance_metrics['avg_performance'] == pytest.approx(0.7) + assert performance_metrics['max_performance'] == 0.85 + assert performance_metrics['min_performance'] == 0.5 + +def test_generate_report(log_directory): + """Test generating a comprehensive log report.""" + parser = LogParser(log_directory) + report = parser.generate_report() + + assert report['total_log_files'] == 3 + assert len(report['file_analyses']) == 3 + + for file_analysis in report['file_analyses']: + assert 'file_name' in file_analysis + assert 'performance' in file_analysis + +def test_invalid_log_file(tmp_path): + """Test handling of invalid log files.""" + invalid_log_dir = tmp_path / "invalid_logs" + invalid_log_dir.mkdir() + + # Create an invalid JSON file + with open(invalid_log_dir / "invalid.json", 'w') as f: + f.write("Not a valid JSON") + + parser = LogParser(str(invalid_log_dir)) + + with pytest.raises(ValueError): + parser.parse_log_file(str(invalid_log_dir / "invalid.json")) + +def test_nonexistent_log_file(): + """Test handling of nonexistent log files.""" + parser = LogParser('/nonexistent/path') + + with pytest.raises(FileNotFoundError): + parser.parse_log_file('/nonexistent/path/log.json') \ No newline at end of file diff --git a/tests/test_iteration_logger.py b/tests/test_iteration_logger.py new file mode 100644 index 0000000..d9da343 --- /dev/null +++ b/tests/test_iteration_logger.py @@ -0,0 +1,135 @@ +import os +import json +import pytest +import logging +from datetime import datetime +from src.iteration_logger import IterationLogger, IterationLogEntry + + +@pytest.fixture +def iteration_logger(tmp_path): + """Create a temporary IterationLogger for testing.""" + log_dir = str(tmp_path) + return IterationLogger(log_dir=log_dir) + + +def test_iteration_log_entry_creation(): + """Test creating an IterationLogEntry.""" + entry = IterationLogEntry( + iteration_number=1, + metadata={'model': 'test_model'}, + performance_metrics={'accuracy': 0.95} + ) + + assert entry.iteration_number == 1 + assert entry.metadata == {'model': 'test_model'} + assert entry.performance_metrics == {'accuracy': 0.95} + assert isinstance(entry.timestamp, datetime) + + +def test_log_iteration(iteration_logger, caplog): + """Test logging an iteration entry.""" + caplog.set_level(logging.INFO) + + entry = IterationLogEntry( + iteration_number=1, + status='success', + performance_metrics={'accuracy': 0.95} + ) + + iteration_logger.log_iteration(entry) + + # Check console log + assert 'Iteration 1 Status: success' in caplog.text + + # Check log file + log_entries = iteration_logger.get_log_entries() + assert len(log_entries) == 1 + assert log_entries[0]['iteration_number'] == 1 + assert log_entries[0]['status'] == 'success' + + +def test_log_multiple_iterations(iteration_logger): + """Test logging multiple iterations.""" + for i in range(3): + entry = IterationLogEntry( + iteration_number=i, + status='success', + performance_metrics={'accuracy': 0.9 + 0.01 * i} + ) + iteration_logger.log_iteration(entry) + + log_entries = iteration_logger.get_log_entries() + assert len(log_entries) == 3 + + # Verify log entries + for i, entry in enumerate(log_entries): + assert entry['iteration_number'] == i + assert entry['status'] == 'success' + + +def test_log_iteration_with_error(iteration_logger): + """Test logging an iteration with error information.""" + entry = IterationLogEntry( + iteration_number=1, + status='failure', + error_info={'type': 'ValueError', 'message': 'Test error'} + ) + + iteration_logger.log_iteration(entry) + + log_entries = iteration_logger.get_log_entries() + assert len(log_entries) == 1 + assert log_entries[0]['error_info'] == {'type': 'ValueError', 'message': 'Test error'} + + +def test_log_entries_serialization(iteration_logger): + """Test JSON serialization of log entries.""" + entry = IterationLogEntry( + iteration_number=1, + metadata={'key': 'value'}, + performance_metrics={'accuracy': 0.95} + ) + + iteration_logger.log_iteration(entry) + + log_entries = iteration_logger.get_log_entries() + log_entry = log_entries[0] + + # Verify timestamp is serialized + assert 'timestamp' in log_entry + assert isinstance(log_entry['timestamp'], str) + + # Verify other fields + assert log_entry['iteration_number'] == 1 + assert log_entry['metadata'] == {'key': 'value'} + assert log_entry['performance_metrics'] == {'accuracy': 0.95} + + +def test_filter_log_entries(iteration_logger): + """Test filtering log entries.""" + # Log multiple entries with different statuses and iteration numbers + entries = [ + IterationLogEntry(iteration_number=1, status='success', performance_metrics={'accuracy': 0.9}), + IterationLogEntry(iteration_number=2, status='failure', performance_metrics={'accuracy': 0.7}), + IterationLogEntry(iteration_number=3, status='success', performance_metrics={'accuracy': 0.95}), + ] + + for entry in entries: + iteration_logger.log_iteration(entry) + + # Test filtering by status + success_entries = iteration_logger.filter_log_entries(status='success') + assert len(success_entries) == 2 + assert all(entry['status'] == 'success' for entry in success_entries) + + # Test filtering by minimum iteration + min_iteration_entries = iteration_logger.filter_log_entries(min_iteration=2) + assert len(min_iteration_entries) == 2 + assert all(entry['iteration_number'] >= 2 for entry in min_iteration_entries) + + # Test combined filtering + combined_filter = iteration_logger.filter_log_entries(status='success', min_iteration=2) + assert len(combined_filter) == 1 + assert combined_filter[0]['iteration_number'] == 3 + assert combined_filter[0]['status'] == 'success' \ No newline at end of file