Update definitions.py

definitions.py

@@ -30,199 +30,199 @@ from pathlib import Path
 # Set logging level from environment variable
 logging.basicConfig(level=os.getenv('LOG_LEVEL', 'INFO'))
 
-    class CodeMetricsAnalyzer:
-        """Analyzes code metrics using various tools"""
+class CodeMetricsAnalyzer:
+    """Analyzes code metrics using various tools"""
 
-        def __init__(self):
-            self.metrics_history = []
-    
-        def analyze_code_quality(self, file_path: str) -> Dict[str, Any]:
-            """Analyzes code quality using multiple metrics"""
-            try:
-                # Pylint analysis
-                pylint_score = self._run_pylint(file_path)
-    
-                # Complexity analysis
-                complexity_score = self._analyze_complexity(file_path)
-
-                # Test coverage analysis
-                coverage_score = self._analyze_test_coverage(file_path)
-                # Security analysis
-                security_score = self._analyze_security(file_path)
-
-                # Calculate overall quality score
-                quality_score = self._calculate_overall_score(
-                    pylint_score,
-                    complexity_score,
-                    coverage_score,
-                    security_score
-                )
-    
-                metrics = {
-                    "quality_score": quality_score,
-                    "pylint_score": pylint_score,
-                    "complexity_score": complexity_score,
-                    "coverage_score": coverage_score,
-                    "security_score": security_score,
-                    "timestamp": datetime.now()
-                }
+    def __init__(self):
+        self.metrics_history = []
 
-                self.metrics_history.append(metrics)
-                return metrics
+    def analyze_code_quality(self, file_path: str) -> Dict[str, Any]:
+        """Analyzes code quality using multiple metrics"""
+        try:
+            # Pylint analysis
+            pylint_score = self._run_pylint(file_path)
+
+            # Complexity analysis
+            complexity_score = self._analyze_complexity(file_path)
+
+            # Test coverage analysis
+            coverage_score = self._analyze_test_coverage(file_path)
+            # Security analysis
+            security_score = self._analyze_security(file_path)
+
+            # Calculate overall quality score
+            quality_score = self._calculate_overall_score(
+                pylint_score,
+                complexity_score,
+                coverage_score,
+                security_score
+            )
 
-            except Exception as e:
-                logging.error(f"Error analyzing code metrics: {str(e)}")
-                return {
-                    "error": str(e),
-                    "quality_score": 0.0,
-                    "timestamp": datetime.now()
-                }
+            metrics = {
+                "quality_score": quality_score,
+                "pylint_score": pylint_score,
+                "complexity_score": complexity_score,
+                "coverage_score": coverage_score,
+                "security_score": security_score,
+                "timestamp": datetime.now()
+            }
 
-        def _run_pylint(self, file_path: str) -> float:
-            """Runs pylint analysis"""
-            try:
-                reporter = JSONReporter()
-                Run([file_path], reporter=reporter, do_exit=False)
-                score = reporter.data.get('score', 0.0)
-                return float(score) / 10.0  # Normalize to 0-1 scale
-            except Exception as e:
-                logging.error(f"Pylint analysis error: {str(e)}")
-                return 0.0
+            self.metrics_history.append(metrics)
+            return metrics
 
-        def _analyze_complexity(self, file_path: str) -> float:
-            """Analyzes code complexity"""
-            try:
-                with open(file_path, 'r') as file:
-                    code = file.read()
-    
-                # Calculate cyclomatic complexity
-                complexity = radon.complexity.cc_visit(code)
-                avg_complexity = sum(item.complexity for item in complexity) / len(complexity) if complexity else 0
+        except Exception as e:
+            logging.error(f"Error analyzing code metrics: {str(e)}")
+            return {
+                "error": str(e),
+                "quality_score": 0.0,
+                "timestamp": datetime.now()
+            }
 
-                # Normalize complexity score (0-1 scale, lower is better)
-                normalized_score = 1.0 - min(avg_complexity / 10.0, 1.0)
-                return normalized_score
-    
-            except Exception as e:
-                logging.error(f"Complexity analysis error: {str(e)}")
-                return 0.0
-    
-        async def _analyze_current_state(self, project_name: str) -> Dict[str, Any]:
-            """Analyze current project state with detailed metrics."""
-            try:
-                self.logger.info(f"Analyzing current state for project: {project_name}")
-    
-                # Collect code metrics
-                code_metrics = await self._collect_code_metrics(project_name)
-                self.logger.info("Code metrics collected successfully.")
-    
-                # Analyze test coverage
-                test_coverage = await self._analyze_test_coverage(project_name)
-                self.logger.info("Test coverage analysis completed.")
-    
-                # Check security vulnerabilities
-                security_analysis = await self._analyze_security(project_name)
-                self.logger.info("Security analysis completed.")
-    
-                # Measure performance metrics
-                performance_metrics = await self._measure_performance(project_name)
-                self.logger.info("Performance metrics measured.")
-    
-                # Determine if requirements are met
-                meets_requirements = await self._check_requirements(
-                    code_metrics,
-                    test_coverage,
-                    security_analysis,
-                    performance_metrics
-                )
-                self.logger.info("Requirements check completed.")
-    
-                return {
-                    "code_metrics": code_metrics,
-                    "test_coverage": test_coverage,
-                    "security_analysis": security_analysis,
-                    "performance_metrics": performance_metrics,
-                    "meets_requirements": meets_requirements,
-                    "timestamp": datetime.now()
-                }
-    
-            except Exception as e:
-                self.logger.error(f"Error analyzing current state: {str(e)}")
-                raise
+    def _run_pylint(self, file_path: str) -> float:
+        """Runs pylint analysis"""
+        try:
+            reporter = JSONReporter()
+            Run([file_path], reporter=reporter, do_exit=False)
+            score = reporter.data.get('score', 0.0)
+            return float(score) / 10.0  # Normalize to 0-1 scale
+        except Exception as e:
+            logging.error(f"Pylint analysis error: {str(e)}")
+            return 0.0
 
-        def _analyze_security(self, file_path: str) -> float:
-            """Analyzes code security using bandit"""
-            try:
-                conf = manager.BanditManager()
-                conf.discover_files([file_path])
-                conf.run_tests()
-    
-                # Calculate security score based on findings
-                total_issues = len(conf.get_issue_list())
-                max_severity = max((issue.severity for issue in conf.get_issue_list()), default=0)
-    
-                # Normalize security score (0-1 scale, higher is better)
-                security_score = 1.0 - (total_issues * max_severity) / 10.0
-                return max(0.0, min(1.0, security_score))
-    
-            except Exception as e:
-                logging.error(f"Security analysis error: {str(e)}")
-                return 0.0
-    
-        def _calculate_overall_score(self, pylint_score: float, complexity_score: float,
-                                   coverage_score: float, security_score: float) -> float:
-            """Calculates overall code quality score"""
-            weights = {
-                'pylint': 0.3,
-                'complexity': 0.2,
-                'coverage': 0.25,
-                'security': 0.25
-            }
-    
-            overall_score = (
-                weights['pylint'] * pylint_score +
-                weights['complexity'] * complexity_score +
-                weights['coverage'] * coverage_score +
-                weights['security'] * security_score
+    def _analyze_complexity(self, file_path: str) -> float:
+        """Analyzes code complexity"""
+        try:
+            with open(file_path, 'r') as file:
+                code = file.read()
+
+            # Calculate cyclomatic complexity
+            complexity = radon.complexity.cc_visit(code)
+            avg_complexity = sum(item.complexity for item in complexity) / len(complexity) if complexity else 0
+
+            # Normalize complexity score (0-1 scale, lower is better)
+            normalized_score = 1.0 - min(avg_complexity / 10.0, 1.0)
+            return normalized_score
+
+        except Exception as e:
+            logging.error(f"Complexity analysis error: {str(e)}")
+            return 0.0
+
+    async def _analyze_current_state(self, project_name: str) -> Dict[str, Any]:
+        """Analyze current project state with detailed metrics."""
+        try:
+            self.logger.info(f"Analyzing current state for project: {project_name}")
+
+            # Collect code metrics
+            code_metrics = await self._collect_code_metrics(project_name)
+            self.logger.info("Code metrics collected successfully.")
+
+            # Analyze test coverage
+            test_coverage = await self._analyze_test_coverage(project_name)
+            self.logger.info("Test coverage analysis completed.")
+
+            # Check security vulnerabilities
+            security_analysis = await self._analyze_security(project_name)
+            self.logger.info("Security analysis completed.")
+
+            # Measure performance metrics
+            performance_metrics = await self._measure_performance(project_name)
+            self.logger.info("Performance metrics measured.")
+
+            # Determine if requirements are met
+            meets_requirements = await self._check_requirements(
+                code_metrics,
+                test_coverage,
+                security_analysis,
+                performance_metrics
             )
+            self.logger.info("Requirements check completed.")
+
+            return {
+                "code_metrics": code_metrics,
+                "test_coverage": test_coverage,
+                "security_analysis": security_analysis,
+                "performance_metrics": performance_metrics,
+                "meets_requirements": meets_requirements,
+                "timestamp": datetime.now()
+            }
 
-            return max(0.0, min(1.0, overall_score))
+        except Exception as e:
+            self.logger.error(f"Error analyzing current state: {str(e)}")
+            raise
+
+    def _analyze_security(self, file_path: str) -> float:
+        """Analyzes code security using bandit"""
+        try:
+            conf = manager.BanditManager()
+            conf.discover_files([file_path])
+            conf.run_tests()
 
-        def get_metrics_history(self) -> List[Dict[str, Any]]:
-            """Returns the history of metrics measurements"""
-            return self.metrics_history
+            # Calculate security score based on findings
+            total_issues = len(conf.get_issue_list())
+            max_severity = max((issue.severity for issue in conf.get_issue_list()), default=0)
 
-    def get_trend_analysis(self) -> Dict[str, Any]:
-        """Analyzes trends in metrics over time"""
-        if not self.metrics_history:
-            return {"status": "No metrics history available"}
+            # Normalize security score (0-1 scale, higher is better)
+            security_score = 1.0 - (total_issues * max_severity) / 10.0
+            return max(0.0, min(1.0, security_score))
 
-        trends = {
-            "quality_score": self._calculate_trend([m["quality_score"] for m in self.metrics_history]),
-            "coverage_score": self._calculate_trend([m["coverage_score"] for m in self.metrics_history]),
-            "security_score": self._calculate_trend([m["security_score"] for m in self.metrics_history])
+        except Exception as e:
+            logging.error(f"Security analysis error: {str(e)}")
+            return 0.0
+
+    def _calculate_overall_score(self, pylint_score: float, complexity_score: float,
+                               coverage_score: float, security_score: float) -> float:
+        """Calculates overall code quality score"""
+        weights = {
+            'pylint': 0.3,
+            'complexity': 0.2,
+            'coverage': 0.25,
+            'security': 0.25
         }
 
-        return trends
+        overall_score = (
+            weights['pylint'] * pylint_score +
+            weights['complexity'] * complexity_score +
+            weights['coverage'] * coverage_score +
+            weights['security'] * security_score
+        )
+
+        return max(0.0, min(1.0, overall_score))
 
-    def _calculate_trend(self, values: List[float]) -> Dict[str, Any]:
-        """Calculates trend statistics for a metric"""
-        if not values:
-            return {"trend": "unknown", "change": 0.0}
+    def get_metrics_history(self) -> List[Dict[str, Any]]:
+        """Returns the history of metrics measurements"""
+        return self.metrics_history
 
-        recent_values = values[-3:]  # Look at last 3 measurements
-        if len(recent_values) < 2:
-            return {"trend": "insufficient data", "change": 0.0}
+def get_trend_analysis(self) -> Dict[str, Any]:
+    """Analyzes trends in metrics over time"""
+    if not self.metrics_history:
+        return {"status": "No metrics history available"}
 
-        change = recent_values[-1] - recent_values[0]
-        trend = "improving" if change > 0 else "declining" if change < 0 else "stable"
+    trends = {
+        "quality_score": self._calculate_trend([m["quality_score"] for m in self.metrics_history]),
+        "coverage_score": self._calculate_trend([m["coverage_score"] for m in self.metrics_history]),
+        "security_score": self._calculate_trend([m["security_score"] for m in self.metrics_history])
+    }
 
-        return {
-            "trend": trend,
-            "change": change,
-            "current": recent_values[-1],
-            "previous": recent_values[0]
-        }
+    return trends
+
+def _calculate_trend(self, values: List[float]) -> Dict[str, Any]:
+    """Calculates trend statistics for a metric"""
+    if not values:
+        return {"trend": "unknown", "change": 0.0}
+
+    recent_values = values[-3:]  # Look at last 3 measurements
+    if len(recent_values) < 2:
+        return {"trend": "insufficient data", "change": 0.0}
+
+    change = recent_values[-1] - recent_values[0]
+    trend = "improving" if change > 0 else "declining" if change < 0 else "stable"
+
+    return {
+        "trend": trend,
+        "change": change,
+        "current": recent_values[-1],
+        "previous": recent_values[0]
+    }
 
 class AutonomousAgentApp:
     """Main application class for the Autonomous Agent System"""