acecalisto3 commited on
Commit
b9c33f3
·
verified ·
1 Parent(s): cddbc9b

Update definitions.py

Browse files
Files changed (1) hide show
  1. definitions.py +1051 -391
definitions.py CHANGED
@@ -1,421 +1,1081 @@
1
- import os
2
- import google.generativeai as genai
3
- import gradio as gr
4
- from gradio.themes.base import Base
5
- from gradio.themes.utils import colors, sizes, fonts
6
- import re
7
  import logging
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8
  import traceback
9
- from typing import List, Dict, AsyncIterator, Optional
10
- from dataclasses import dataclass, field
11
-
12
- # --- Logging Setup ---
13
- logging.basicConfig(level=logging.INFO)
14
-
15
- # --- Custom Exception ---
16
- class APIConfigError(Exception):
17
- """Custom exception for API configuration issues."""
18
- pass
19
-
20
- # --- Data Classes ---
21
- @dataclass
22
- class ChatMessage:
23
- role: str
24
- content: str
25
-
26
- @dataclass
27
- class DevelopmentContext:
28
- user_task: str
29
- relevant_code: Optional[str] = None
30
- tech_stack: List[str] = field(default_factory=list)
31
- user_intent: Optional[str] = None
32
- code_language: Optional[str] = None
33
-
34
- # --- GeminiDevAssistant Class ---
35
- class GeminiDevAssistant:
36
- def __init__(self, api_key: str):
37
  try:
38
- self.configure_api(api_key)
39
- self.model = genai.GenerativeModel('gemini-1.5-flash-002')
40
- logging.info("GeminiDevAssistant initialized successfully.")
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
41
  except Exception as e:
42
- logging.error(f"Error initializing Gemini API: {e}")
43
- raise APIConfigError(f"Error initializing Gemini API: {e}")
44
-
45
- @staticmethod
46
- def configure_api(api_key: str) -> None:
47
- genai.configure(api_key=api_key)
48
-
49
- def analyze_development_task(self, user_message: str) -> DevelopmentContext:
50
- """Analyzes the user's development task and extracts relevant context."""
51
- logging.info("Analyzing development task...")
52
- code_blocks = self._extract_code_blocks(user_message)
53
- code = code_blocks[0]['code'] if code_blocks else None
54
- language = code_blocks[0]['language'] if code_blocks else None
55
-
56
- intent = self._infer_user_intent(user_message)
57
- tech_stack = self._detect_tech_stack(user_message, code, language)
58
-
59
- logging.debug(f"Detected code: {code}")
60
- logging.debug(f"Detected language: {language}")
61
- logging.debug(f"Inferred intent: {intent}")
62
- logging.debug(f"Detected tech stack: {tech_stack}")
63
-
64
- return DevelopmentContext(
65
- user_task=user_message,
66
- relevant_code=code,
67
- tech_stack=tech_stack,
68
- user_intent=intent,
69
- code_language=language
70
- )
71
 
72
- def _extract_code_blocks(self, text: str) -> List[Dict[str, str]]:
73
- """Extracts code blocks from text using regex."""
74
- code_blocks = []
75
- matches = re.findall(r"```(\w+)?\n(.*?)\n```", text, re.DOTALL)
76
- for match in matches:
77
- language = match[0] if match[0] else "plaintext"
78
- code = match[1].strip()
79
- code_blocks.append({"language": language, "code": code})
80
- return code_blocks
81
-
82
- def _infer_user_intent(self, user_message: str) -> Optional[str]:
83
- """Basic intent inference based on keywords."""
84
- message_lower = user_message.lower()
85
- if "debug" in message_lower or "error" in message_lower or "fix" in message_lower:
86
- return "debugging"
87
- elif "optimize" in message_lower or "performance" in message_lower or "improve" in message_lower:
88
- return "optimization"
89
- elif "test" in message_lower or "unit test" in message_lower:
90
- return "unit testing"
91
- elif "refactor" in message_lower or "clean up" in message_lower:
92
- return "refactoring"
93
- elif "document" in message_lower or "comment" in message_lower:
94
- return "documentation elif "generate" in message_lower or "create" in message_lower or "write" in message_lower:
95
- return "code generation"
96
- return "general development task"
97
-
98
- def _detect_tech_stack(self, user_message: str, code: Optional[str], language: Optional[str]) -> List[str]:
99
- """Basic tech stack detection based on keywords and code analysis."""
100
- tech_stack = set()
101
- message_lower = user_message.lower()
102
-
103
- if language:
104
- tech_stack.add(language)
105
-
106
- common_tech_keywords = {
107
- "python": ["python", "django", "flask", "pandas", "numpy"],
108
- "javascript": ["javascript", "react", "vue", "angular", "node.js", "express"],
109
- "java": ["java", "spring", "hibernate"],
110
- "c++": ["c++", "qt"],
111
- "c#": ["c#", ".net", "asp.net"],
112
- "typescript": ["typescript", "react", "angular"],
113
- "html": ["html", "css", "javascript"],
114
- "css": ["css", "html", "javascript"],
115
- "sql": ["sql", "mysql", "postgresql", "sqlite", "database"],
116
- "docker": ["docker", "container", "dockerfile"],
117
- "kubernetes": ["kubernetes", "k8s", "cluster"],
118
- "git": ["git", "version control", "repository"],
119
- "aws": ["aws", "amazon web services", "ec2", "s3", "lambda"],
120
- "azure": ["azure", "microsoft azure", "cloud services"],
121
- "gcp": ["gcp", "google cloud platform", "google cloud"],
122
- }
123
 
124
- for tech, keywords in common_tech_keywords.items():
125
- for keyword in keywords:
126
- if keyword in message_lower or (code and keyword in code.lower()):
127
- tech_stack.add(tech)
128
- break
129
-
130
- return list(tech_stack)
131
-
132
- def format_chat_history(self, messages: List[Dict]) -> List[Dict]:
133
- """Formats chat history for Gemini API."""
134
- formatted_history = []
135
- for msg in messages:
136
- if isinstance(msg, dict) and "role" in msg and "content" in msg:
137
- role = msg["role"]
138
- content = msg["content"]
139
- formatted_history.append({"role": role, "parts": [content]})
140
- else:
141
- logging.warning(f"Unexpected message format: {msg}")
142
- return formatted_history
143
 
144
- async def stream_development_response(self, user_message: str, messages: List[Dict]) -> AsyncIterator[List[Dict]]:
145
- """Streams development responses from Gemini API (Synchronous Iteration)."""
146
- development_context = self.analyze_development_task(user_message)
147
- prompt = self._build_development_prompt(development_context)
148
- chat_history = self.format_chat_history(messages)
149
 
150
- logging.info("Sending prompt to Gemini API...")
151
- logging.debug(f"Prompt being sent: {prompt}")
 
152
 
 
 
153
  try:
154
- chat = self.model.start_chat(history=chat_history)
155
- response = chat.send_message(prompt, stream=True)
156
 
157
- response_iterator = iter(response)
 
 
158
 
159
- try:
160
- while True:
161
- chunk = next(response_iterator)
162
 
163
- updated_messages, code_block_buffer = self._process_development_stream(
164
- chunk, messages, "", "", False, "", development_context.code_language
165
- )
166
- gradio_messages_format = [(msg.role, self._format_content_for_gradio(msg.content, development_context.code_language)) if msg.role == 'user' else (None, self._format_content_for_gradio(msg.content, development_context.code_language)) for msg in updated_messages]
167
- yield gradio_messages_format
168
 
169
- except StopIteration:
170
- pass
171
- except Exception as e:
172
- error_message = self._format_development_error(e)
173
- logging.error(f"Error during Gemini API call (in loop): {e}\n{traceback.format_exc()}")
174
- yield messages + [(None, error_message)]
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
175
 
176
  except Exception as e:
177
- error_message = self._format_development_error(e)
178
- logging.error(f"Error during Gemini API call (outer): {e}\n{traceback.format_exc()}")
179
- yield messages + [(None, error_message)]
180
-
181
- def _process_development_stream(self, chunk, messages, thought_buffer, response_buffer, thinking_complete, code_block_buffer, code_language):
182
- """Processes a chunk of the development stream and updates messages, handling code blocks."""
183
- text_chunk = chunk.text
184
-
185
- response_buffer += text_chunk
186
-
187
- if "```" in response_buffer:
188
- parts = response_buffer.split("```")
189
- response_buffer = parts[-1]
190
- code_block = "```" + parts[-2] + "```" if len(parts) > 1 else ""
191
- if code_block:
192
- code_block_buffer += code_block
193
-
194
- if messages and messages[-1].role == "assistant":
195
- messages[-1].content = response_buffer
196
- if code_block_buffer:
197
- messages [-1].content = code_block_buffer + "\n" + messages[-1].content
198
- code_block_buffer = ""
199
- elif messages:
200
- messages.append(ChatMessage(role="assistant", content=response_buffer))
201
- else:
202
- messages.append(ChatMessage(role="assistant", content=response_buffer))
203
-
204
- return messages, code_block_buffer
205
-
206
- def _format_content_for_gradio(self, content: str, code_language: Optional[str]) -> str:
207
- """Formats content for Gradio chatbot, adding syntax highlighting for code blocks."""
208
- formatted_content = content
209
- code_blocks = self._extract_code_blocks(content)
210
-
211
- if code_blocks:
212
- for block in code_blocks:
213
- lang = block['language']
214
- code = block['code']
215
- formatted_code = f"```{lang}\n{code}\n```"
216
- formatted_content = formatted_content.replace(f"```{lang}\n{code}\n```", formatted_code)
217
-
218
- return formatted_content
219
-
220
- def _build_development_prompt(self, development_context: DevelopmentContext) -> str:
221
- """Builds a more detailed and effective prompt for the Gemini model."""
222
- prompt_parts = [
223
- "You are an expert Development Assistant. Your goal is to provide the most helpful and effective assistance to users with their development tasks.",
224
- "Focus on providing practical solutions, code examples, debugging guidance, optimization suggestions, and best practices.",
225
- "When providing code, ensure it is correct, efficient, and well-formatted. Use markdown code blocks with appropriate language syntax highlighting.",
226
- "Clearly explain your reasoning and thought process step-by-step, especially for complex tasks like debugging or optimization.",
227
- "If the user provides code, analyze it carefully and provide specific feedback related to their task.",
228
- f"**User 's Development Task:** {development_context.user_task}.",
229
- ]
230
-
231
- if development_context.relevant_code:
232
- prompt_parts.append(f"\n**Relevant Code Snippet (Language: {development_context.code_language or 'unknown'}):**\n```{development_context.code_language or 'plaintext'}\n{development_context.relevant_code}\n```")
233
-
234
- if development_context.tech_stack:
235
- prompt_parts.append(f"\n**Detected Technologies/Tech Stack:** {', '.join(development_context.tech_stack)}.")
236
-
237
- if development_context.user_intent:
238
- prompt_parts.append(f"\n**Inferred User Intent:** User seems to be asking for help with {development_context.user_intent}.")
239
-
240
- prompt_parts.append(
241
- "\n**Instructions for Response Formatting:**\n"
242
- "1. Start with a brief summary of your understanding of the user's task and intent.\n"
243
- "2. If providing code, use markdown code blocks with language syntax highlighting.\n"
244
- "3. Clearly separate your thought process, explanations, and code examples.\n"
245
- "4. Be concise but comprehensive. Avoid unnecessary verbosity.\n"
246
- "5. If you identify errors or issues, clearly point them out and suggest solutions.\n"
247
- "6. If suggesting optimizations, explain the benefits and trade-offs.\n"
248
- "7. End with a helpful concluding remark or next steps for the user."
249
  )
250
 
251
- return " ".join(prompt_parts)
252
-
253
- def _format_development_error(self, error: Exception) -> str:
254
- """Formats an enhanced error message for display to the user."""
255
- error_type = type(error).__name__
256
- error_details = str(error)
257
- full_traceback = traceback.format_exc()
258
-
259
- error_message = f"**Oops! Development Task Encountered an Error:**\n\n"
260
- error_message += f"**Error Type:** `{error_type}`\n"
261
- error_message += f"**Details:** {error_details}\n\n"
262
- error_message += "**Detailed Error Traceback (for debugging):**\n```\n{full_traceback}\n```\n\n"
263
- error_message += "**Suggestions to Troubleshoot:**\n"
264
- error_message += "* **Check your Gemini API key:** Ensure it's correctly set as an environment variable `GEMINI_API_KEY`.\n"
265
- error_message += "* **Verify your network connection:** Make sure you have internet access.\n"
266
- error_message += "* **Simplify your request:** Try breaking down complex tasks into smaller, more manageable steps.\n"
267
- error_message += "* **Provide more context:** The more information you give, the better the assistant can understand your request.\n"
268
- error_message += "* **Review the error traceback:** The 'Detailed Error Traceback' above might contain clues about the underlying issue. If you're a developer, this is often the most helpful part.\n"
269
-
270
- logging.error(f"Formatted Error Message: {error_message}")
271
-
272
- return error_message
273
-
274
- # --- Gradio Interface ---
275
- def create_dev_assistant_interface():
276
- try:
277
- api_key = os.getenv("GEMINI_API_KEY")
278
- if not api_key:
279
- raise APIConfigError("GEMINI_API_KEY environment variable not set. Please set your Gemini API key.")
280
-
281
- assistant = Gemini DevAssistant(api_key)
282
-
283
- async def process_input(msg, history_gradio):
284
- history = []
285
- for human, assistant_response in history_gradio:
286
- if human is not None:
287
- history.append({"role": "user", "content": human})
288
- if assistant_response is not None:
289
- history.append({"role": "model", "content": assistant_response})
290
-
291
- async def stream_output():
292
- async for messages_list in assistant.stream_development_response(msg, history):
293
- yield messages_list
294
-
295
- collected_messages = []
296
- async for message_chunk in stream_output():
297
- collected_messages = message_chunk
298
- yield collected_messages
299
-
300
- # --- Custom Theme ---
301
- class ModernDevTheme(Base):
302
- def __init__(self):
303
- super().__init__(
304
- primary_hue="indigo",
305
- secondary_hue="blue",
306
- neutral_hue="slate",
307
- font=fonts.GoogleFont("Inter"),
308
- font_mono=fonts.GoogleFont("JetBrains Mono")
309
- )
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
310
 
311
- # Enhanced colors
312
- self.colors = {
313
- "background": {
314
- "primary": "#ffffff",
315
- "secondary": "#f8fafc",
316
- "tertiary": "#f1f5f9"
317
- },
318
- "border": "#e2e8f0",
319
- "text": {
320
- "primary": "#1e293b",
321
- "secondary": "#64748b"
322
- },
323
- "primary": "#4f46e5",
324
- "secondary": "#3b82f6",
325
- "success": "#22c55e",
326
- "error": "#ef4444"
327
- }
328
 
329
- # Enhanced components
330
- self.spacing = sizes.spacing_md
331
- self.radius = sizes.radius_md
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
332
 
333
- # Component styles
334
- self.button_primary = {
335
- "background": self.colors["primary"],
336
- "text_color": "#ffffff",
337
- "border": "none",
338
- "padding": "12px 20px",
339
- "border_radius": "8px",
340
- "font_weight": "600",
341
- "hover": {"opacity": 0.9}
342
- }
343
 
344
- self.input = {
345
- "background": "#ffffff",
346
- "border": f"1px solid {self.colors['border']}",
347
- "border_radius": "8px",
348
- "padding": "12px 16px"
349
- }
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
350
 
351
- # --- UI Layout Function ---
352
- def create_ui(assistant):
353
- with gr.Blocks(theme=ModernDevTheme()) as interface:
354
- gr.Markdown(
355
- """
356
- # AI Development Assistant
357
- Get expert help with coding, debugging, and software development
358
- """
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
359
  )
360
 
361
- with gr.Row():
362
- with gr.Column(scale=3):
363
- chatbot = gr.Chatbot(
364
- height=600,
365
- show_label=False,
366
- container=True,
367
- bubble_full_width=False,
368
- show_copy_button=True,
369
- avatar_images=("👤", "🤖")
370
- )
371
-
372
- with gr.Row():
373
- msg = gr.Textbox(
374
- placeholder="Ask your development question...",
375
- show_label=False,
376
- container=False,
377
- scale=12
378
- )
379
- submit = gr.Button(
380
- "Send",
381
- variant="primary",
382
- scale=1
383
- )
384
-
385
- with gr.Column(scale=1):
386
- gr.Markdown("### Development Context")
387
- context_info = gr.JSON(
388
- label="Current Task Analysis",
389
- value={}
390
- )
391
 
392
- # Event handlers remain the same
393
- msg.submit(assistant.stream_development_response)
394
- submit.click(assistant.stream_development_response)
395
-
396
- return interface
397
-
398
- except APIConfigError as e:
399
- return gr.Interface(
400
- fn=lambda: [None],
401
- inputs=None,
402
- outputs=gr.Chatbot(label="Development Assistant"),
403
- title="System Initialization Error",
404
- description=str(e)
 
 
 
 
 
 
 
 
 
 
 
 
405
  )
406
- except Exception as e:
407
- return gr.Interface(
408
- fn=lambda: [None],
409
- inputs=None,
410
- outputs=gr.Chatbot(label="Development Assistant"),
411
- title="Interface Malfunction",
412
- description=f"System interface encountered an unexpected malfunction: {str(e)}"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
413
  )
414
 
415
- # --- Main execution ---
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
416
  if __name__ == "__main__":
417
- demo = create_dev_assistant_interface()
418
- if demo:
419
- demo.launch()
420
- else:
421
- print("Gradio demo could not be initialized due to errors.")
 
1
+ import platform
2
+ import streamlit as st
3
+ import psutil
4
+ from typing import List, Dict, Optional, Any, Tuple
5
+ from dataclasses import dataclass
6
+ from enum import Enum
7
  import logging
8
+ import time
9
+ import ast
10
+ import pylint.lint
11
+ import radon.complexity
12
+ import radon.metrics
13
+ from pylint.lint import Run
14
+ from pylint.reporters import JSONReporter
15
+ from coverage import Coverage
16
+ import bandit
17
+ from bandit.core import manager
18
+ from datetime import datetime
19
+ import os
20
+ import sys
21
+ import requests
22
+ import asyncio
23
+ import statistics
24
+ import json
25
  import traceback
26
+ from pathlib import Path
27
+
28
+ # Set logging level from environment variable
29
+ logging.basicConfig(level=os.getenv('LOG_LEVEL', 'INFO'))
30
+
31
+ class CodeMetricsAnalyzer:
32
+ """Analyzes code metrics using various tools"""
33
+
34
+ def __init__(self):
35
+ self.metrics_history = []
36
+
37
+ def analyze_code_quality(self, file_path: str) -> Dict[str, Any]:
38
+ """Analyzes code quality using multiple metrics"""
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
39
  try:
40
+ # Pylint analysis
41
+ pylint_score = self._run_pylint(file_path)
42
+
43
+ # Complexity analysis
44
+ complexity_score = self._analyze_complexity(file_path)
45
+
46
+ # Test coverage analysis
47
+ coverage_score = self._analyze_test_coverage(file_path)
48
+
49
+ # Security analysis
50
+ security_score = self._analyze_security(file_path)
51
+
52
+ # Calculate overall quality score
53
+ quality_score = self._calculate_overall_score(
54
+ pylint_score,
55
+ complexity_score,
56
+ coverage_score,
57
+ security_score
58
+ )
59
+
60
+ metrics = {
61
+ "quality_score": quality_score,
62
+ "pylint_score": pylint_score,
63
+ "complexity_score": complexity_score,
64
+ "coverage_score": coverage_score,
65
+ "security_score": security_score,
66
+ "timestamp": datetime.now()
67
+ }
68
+
69
+ self.metrics_history.append(metrics)
70
+ return metrics
71
+
72
  except Exception as e:
73
+ logging.error(f"Error analyzing code metrics: {str(e)}")
74
+ return {
75
+ "error": str(e),
76
+ "quality_score": 0.0,
77
+ "timestamp": datetime.now()
78
+ }
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
79
 
80
+ def _run_pylint(self, file_path: str) -> float:
81
+ """Runs pylint analysis"""
82
+ try:
83
+ reporter = JSONReporter()
84
+ Run([file_path], reporter=reporter, do_exit=False)
85
+ score = reporter.data.get('score', 0.0)
86
+ return float(score) / 10.0 # Normalize to 0-1 scale
87
+ except Exception as e:
88
+ logging.error(f"Pylint analysis error: {str(e)}")
89
+ return 0.0
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
90
 
91
+ def _analyze_complexity(self, file_path: str) -> float:
92
+ """Analyzes code complexity"""
93
+ try:
94
+ with open(file_path, 'r') as file:
95
+ code = file.read()
96
+
97
+ # Calculate cyclomatic complexity
98
+ complexity = radon.complexity.cc_visit(code)
99
+ avg_complexity = sum(item.complexity for item in complexity) / len(complexity) if complexity else 0
 
 
 
 
 
 
 
 
 
 
100
 
101
+ # Normalize complexity score (0-1 scale, lower is better)
102
+ normalized_score = 1.0 - min(avg_complexity / 10.0, 1.0)
103
+ return normalized_score
 
 
104
 
105
+ except Exception as e:
106
+ logging.error(f"Complexity analysis error: {str(e)}")
107
+ return 0.0
108
 
109
+ async def _analyze_current_state(self, project_name: str) -> Dict[str, Any]:
110
+ """Analyze current project state with detailed metrics."""
111
  try:
112
+ self.logger.info(f"Analyzing current state for project: {project_name}")
 
113
 
114
+ # Collect code metrics
115
+ code_metrics = await self._collect_code_metrics(project_name)
116
+ self.logger.info("Code metrics collected successfully.")
117
 
118
+ # Analyze test coverage
119
+ test_coverage = await self._analyze_test_coverage(project_name)
120
+ self.logger.info("Test coverage analysis completed.")
121
 
122
+ # Check security vulnerabilities
123
+ security_analysis = await self._analyze_security(project_name)
124
+ self.logger.info("Security analysis completed.")
 
 
125
 
126
+ # Measure performance metrics
127
+ performance_metrics = await self._measure_performance(project_name)
128
+ self.logger.info("Performance metrics measured.")
129
+
130
+ # Determine if requirements are met
131
+ meets_requirements = await self._check_requirements(
132
+ code_metrics,
133
+ test_coverage,
134
+ security_analysis,
135
+ performance_metrics
136
+ )
137
+ self.logger.info("Requirements check completed.")
138
+
139
+ return {
140
+ "code_metrics": code_metrics,
141
+ "test_coverage": test_coverage,
142
+ "security_analysis": security_analysis,
143
+ "performance_metrics": performance_metrics,
144
+ "meets_requirements": meets_requirements,
145
+ "timestamp": datetime.now()
146
+ }
147
 
148
  except Exception as e:
149
+ self.logger.error(f"Error analyzing current state: {str(e)}")
150
+ raise
151
+
152
+ def _analyze_security (file_path: str) -> float:
153
+ """Analyzes code security using bandit"""
154
+ try:
155
+ conf = manager.BanditManager()
156
+ conf.discover_files([file_path])
157
+ conf.run_tests()
158
+
159
+ # Calculate security score based on findings
160
+ total_issues = len(conf.get_issue_list())
161
+ max_severity = max((issue.severity for issue in conf.get_issue_list()), default=0)
162
+
163
+ # Normalize security score (0-1 scale, higher is better)
164
+ security_score = 1.0 - (total_issues * max_severity) / 10.0
165
+ return max(0.0, min(1.0, security_score))
166
+
167
+ except Exception as e:
168
+ logging.error(f"Security analysis error: {str(e)}")
169
+ return 0.0
170
+
171
+ def _calculate_overall_score(self, pylint_score: float, complexity_score: float,
172
+ coverage_score: float, security_score: float) -> float:
173
+ """Calculates overall code quality score"""
174
+ weights = {
175
+ 'pylint': 0.3,
176
+ 'complexity': 0.2,
177
+ 'coverage': 0.25,
178
+ 'security': 0.25
179
+ }
180
+
181
+ overall_score = (
182
+ weights['pylint'] * pylint_score +
183
+ weights['complexity'] * complexity_score +
184
+ weights['coverage'] * coverage_score +
185
+ weights['security'] * security_score
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
186
  )
187
 
188
+ return max(0.0, min(1.0, overall_score))
189
+
190
+ def get_metrics_history(self) -> List[Dict[str, Any]]:
191
+ """Returns the history of metrics measurements"""
192
+ return self.metrics_history
193
+
194
+ def get_trend_analysis(self) -> Dict[str, Any]:
195
+ """Analyzes trends in metrics over time"""
196
+ if not self.metrics_history:
197
+ return {"status": "No metrics history available"}
198
+
199
+ trends = {
200
+ "quality_score": self._calculate_trend([m["quality_score"] for m in self.metrics_history]),
201
+ "coverage_score": self._calculate_trend([m["coverage_score"] for m in self.metrics_history]),
202
+ "security_score": self._calculate_trend([m["security_score"] for m in self.metrics_history])
203
+ }
204
+
205
+ return trends
206
+
207
+ def _calculate_trend(self, values: List[float]) -> Dict[str, Any]:
208
+ """Calculates trend statistics for a metric"""
209
+ if not values:
210
+ return {"trend": "unknown", "change": 0.0}
211
+
212
+ recent_values = values[-3:] # Look at last 3 measurements
213
+ if len(recent_values) < 2:
214
+ return {"trend": "insufficient data", "change": 0.0}
215
+
216
+ change = recent_values[-1] - recent_values[0]
217
+ trend = "improving" if change > 0 else "declining" if change < 0 else "stable"
218
+
219
+ return {
220
+ "trend": trend,
221
+ "change": change,
222
+ "current": recent_values[-1],
223
+ "previous": recent_values[0]
224
+ }
225
+
226
+ class WorkspaceManager:
227
+ """Manages the workspace for the Autonomous Agent System."""
228
+
229
+ def __init__(self, workspace_dir: str):
230
+ self.workspace_dir = workspace_dir
231
+
232
+ def get_workspace_tree(self) -> Dict[str, Any]:
233
+ """Get the structure of the workspace."""
234
+ # Placeholder implementation
235
+ return {"workspace": "tree_structure"}
236
+
237
+ def create_file(self, filename: str, content: str) -> str:
238
+ """Create a new file in the workspace."""
239
+ file_path = os.path.join(self.workspace_dir, filename)
240
+ with open(file_path, 'w') as file:
241
+ file.write(content)
242
+ return f"File '{filename}' created successfully."
243
+
244
+ def delete_file(self, filename: str) -> str:
245
+ """Delete a file from the workspace."""
246
+ file_path = os.path.join(self.workspace_dir, filename)
247
+ if os.path.exists(file_path):
248
+ os.remove(file_path)
249
+ return f"File '{filename}' deleted successfully."
250
+ return f"File '{filename}' not found."
251
+
252
+ class ToolManager:
253
+ """Manages tools for the autonomous agent system."""
254
+
255
+ def __init__(self):
256
+ self.tools = {}
257
+
258
+ def add_tool(self, tool_name, tool_config):
259
+ """Add a tool to the tool manager."""
260
+ self.tools[tool_name] = tool_config
261
+
262
+ def get_tool(self, tool_name):
263
+ """Get a tool from the tool manager."""
264
+ return self.tools.get(tool_name)
265
+
266
+ def remove_tool(self, tool_name):
267
+ """Remove a tool from the tool manager."""
268
+ if tool_name in self.tools:
269
+ del self.tools[tool_name]
270
+
271
+ class QualityMetrics:
272
+ """Advanced quality metrics tracking and analysis"""
273
+ def __init__(self):
274
+ self.metrics_analyzer = CodeMetricsAnalyzer()
275
+ self.code_quality_score = 0.0
276
+ self.test_coverage = 0.0
277
+ self.security_score = "unknown"
278
+ self.performance_score = 0.0
279
+ self.history = []
280
+ self.thresholds = {
281
+ "code_quality": 0.85,
282
+ "test_coverage": 0.90,
283
+ "security": 0.85,
284
+ "performance": 0.80
285
+ }
286
+
287
+ class AutonomousAgent:
288
+ """Autonomous agent for the system."""
289
+ def __init__(self):
290
+ self.chat_system = None
291
+ self.refinement_loop = None
292
+ self.tool_repository = None
293
+
294
+ def initialize_tool_repository(self, tool_repository):
295
+ """Initializes the tool repository."""
296
+ self.tool_repository = tool_repository
297
+
298
+ def build_tool(self, tool_name, task):
299
+ """Builds a tool."""
300
+ tool = self.tool_repository.get_tool(tool_name)
301
+ if tool:
302
+ tool.run(task)
303
+ return f"{tool_name} built and ran successfully."
304
+ else:
305
+ return f"{tool_name} not found in tool repository."
306
+
307
+ def build_agent(self, agent_name, role):
308
+ """Builds an agent."""
309
+ agent = self._create_agent(agent_name, role)
310
+ if agent:
311
+ return f"{agent_name} agent built successfully."
312
+ else:
313
+ return f"{agent_name} agent creation failed."
314
+
315
+ def _create_agent(self, agent_name, role):
316
+ """Creates a new agent."""
317
+ if role == "development":
318
+ return self.DevelopmentAgent(agent_name)
319
+ elif role == "testing":
320
+ return self.TestingAgent(agent_name)
321
+ elif role == "security":
322
+ return self.SecurityAgent(agent_name)
323
+ else:
324
+ return None
325
+
326
+ class DevelopmentAgent:
327
+ """Development agent."""
328
+ def __init__(self, name):
329
+ self.name = name
330
+
331
+ def build_tool(self, tool_name, task):
332
+ """Builds a development tool."""
333
+ return f"{self.name} built {tool_name} development tool."
334
+
335
+ class TestingAgent:
336
+ """Testing agent."""
337
+ def __init__(self, name):
338
+ self.name = name
339
+
340
+ def build_tool(self, tool_name, task):
341
+ """Builds a testing tool."""
342
+ return f"{self.name} built {tool_name} testing tool."
343
+
344
+ class SecurityAgent:
345
+ """Security agent."""
346
+ def __init__(self, name):
347
+ self.name = name
348
+
349
+ def build_tool(self, tool_name, task):
350
+ """Builds a security tool."""
351
+ return f"{self.name} built {tool_name} security tool."
352
+
353
+ class ToolRepository:
354
+ """Tool repository."""
355
+ def __init__(self):
356
+ self.tools = []
357
+
358
+ def add_tool(self, tool):
359
+ """Adds a tool to the repository."""
360
+ self.tools.append(tool)
361
+
362
+ def get_tool(self, tool_name):
363
+ """Gets a tool from the repository."""
364
+ for tool in self.tools:
365
+ if tool.name == tool_name:
366
+ return tool
367
+
368
+ class Tool:
369
+ """Tool."""
370
+ def __init__(self, name, task):
371
+ self.name = name
372
+ self.task = task
373
+
374
+ def run(self, task):
375
+ """Runs the tool."""
376
+ print(f"Running {self.name} tool for {task}")
377
+
378
+ class ChatSystem:
379
+ """Manages the chat interaction between users and the autonomous system."""
380
+ def __init__(self, agent: 'AutonomousAgentApp.AutonomousAgent'):
381
+ self.agent = agent
382
+ self.chat_history = []
383
+ self.active_tasks = {}
384
+ self.command_handlers = {
385
+ '/task': self.handle_task_command,
386
+ '/status': self.handle_status_command,
387
+ '/stop': self.handle_stop_command,
388
+ '/help': self.handle_help_command,
389
+ '/modify': self.handle_modify_command
390
+ }
391
+
392
+ def render_chat_interface(self):
393
+ """Render the chat interface in Streamlit sidebar."""
394
+ with st.sidebar:
395
+ st.markdown("---")
396
+ st.subheader("System Chat")
397
 
398
+ # Chat controls
399
+ if st.button("Clear Chat History"):
400
+ self.clear_chat_history()
401
+
402
+ # Chat history display
403
+ chat_container = st.container()
404
+ with chat_container:
405
+ for message in self.chat_history:
406
+ self._render_message(message)
407
+
408
+ # Input area
409
+ user_input = st.text_input("Type message/command...", key="chat_input")
410
+ if st.button("Send", key="send_message"):
411
+ self.process_user_input(user_input)
 
 
 
412
 
413
+ class RefinementLoop:
414
+ """Manages the iterative refinement process."""
415
+ def __init__(self, pipeline):
416
+ self.pipeline = pipeline
417
+ self.max_iterations = 10
418
+ self.quality_metrics = QualityMetrics()
419
+ self.logger = logging.getLogger(__name__)
420
+ self.current_iteration = 0
421
+ self.history = []
422
+
423
+ async def run_refinement_cycle(self, task: str) -> Dict[str, Any]:
424
+ """Run a complete refinement cycle for the given task."""
425
+ self.logger.info(f"Starting refinement cycle for task: {task}")
426
+ self.current_iteration = 0
427
+
428
+ try:
429
+ while self.current_iteration < self.max_iterations:
430
+ self.logger.info(f"Starting iteration {self.current_iteration + 1}")
431
 
432
+ # Execute pipeline stages
433
+ planning_result = await self.pipeline.execute_stage(
434
+ self.pipeline.PipelineStage.PLANNING,
435
+ {"task": task}
436
+ )
 
 
 
 
 
437
 
438
+ development_result = await self.pipeline.execute_stage(
439
+ self.pipeline.PipelineStage.DEVELOPMENT,
440
+ planning_result["result"]
441
+ )
442
+
443
+ testing_result = await self.pipeline.execute_stage(
444
+ self.pipeline.PipelineStage.TESTING,
445
+ development_result["result"]
446
+ )
447
+
448
+ # Analyze results
449
+ quality_analysis = self._analyze_quality(testing_result["result"])
450
+
451
+ # Record iteration history
452
+ self.history.append({
453
+ "iteration": self.current_iteration,
454
+ "quality_metrics": quality_analysis,
455
+ "timestamp": datetime.now()
456
+ })
457
+
458
+ # Check if quality requirements are met
459
+ if self._meets_quality_requirements(quality_analysis):
460
+ self.logger.info("Quality requirements met. Refinement cycle complete.")
461
+ return self._prepare_final_result(quality_analysis)
462
+
463
+ self.current_iteration += 1
464
+
465
+ return {
466
+ "status": "max_iterations_reached",
467
+ "iterations_completed": self.current_iteration,
468
+ "final_quality": quality_analysis
469
+ }
470
+
471
+ except Exception as e:
472
+ self.logger.error(f"Error in refinement cycle: {str(e)}")
473
+ return {"status": "error", "error": str(e)}
474
+
475
+ def _analyze_quality(self, result: Dict[str, Any]) -> Dict[str, float]:
476
+ """Analyze the quality metrics of the current iteration."""
477
+ return {
478
+ "code_quality": self.quality_metrics.code_quality_score,
479
+ "test_coverage": self.quality_metrics.test_coverage,
480
+ "security_score": float(self.quality_metrics.security_score)
481
+ }
482
+
483
+ def _meets_quality_requirements(self, quality_analysis: Dict[str, float]) -> bool:
484
+ """Check if the current quality metrics meet the requirements."""
485
+ thresholds = self.quality_metrics.thresholds
486
+ return (
487
+ quality_analysis["code_quality"] >= thresholds["code_quality"] and
488
+ quality_analysis["test_coverage"] >= thresholds["test_coverage"] and
489
+ quality_analysis["security_score"] >= thresholds["security"]
490
+ )
491
+
492
+ def _prepare_final_result(self, quality_analysis: Dict[str, float]) -> Dict[str, Any]:
493
+ """Prepare the final result of the refinement cycle."""
494
+ return {
495
+ "status": "success",
496
+ "iterations_completed": self.current_iteration,
497
+ "final_quality": quality_analysis,
498
+ "history": self.history
499
+ }
500
+
501
+ def get_refinement_history(self) -> List[Dict[str, Any]]:
502
+ """Get the history of refinement iterations."""
503
+ return self.history
504
+
505
+ class AutonomousAgentApp:
506
+ """Main application class for the Autonomous Agent System."""
507
+ def __init__(self):
508
+ self.workspace_manager = WorkspaceManager(workspace_dir=os.getenv('WORKSPACE_DIR', 'workspace'))
509
+ self.pipeline = self._initialize_pipeline()
510
+ self.refinement_loop = self.RefinementLoop(pipeline=self.pipeline) # Corrected line
511
+ self.interface = self.StreamlitInterface(self)
512
+ self.autonomous_agent = self.AutonomousAgent()
513
+ self.tools_repository = self._initialize_tool_repository()
514
+ self.autonomous_agent.initialize_tool_repository(self.tools_repository)
515
+
516
+ def _initialize_tool_repository(self):
517
+ """Initializes the tool repository."""
518
+ tool_repository = ToolRepository()
519
+ tool_repository.add_tool(Tool(name="Code Review Tool", task="Review Code Quality"))
520
+ tool_repository.add_tool(Tool(name="Automated Testing Tool", task="Run Automated Tests"))
521
+ tool_repository.add_tool(Tool(name="Security Audit Tool", task="Run Security Audits"))
522
+ return tool_repository
523
+
524
+ def _initialize_pipeline(self) -> 'AutonomousAgentApp.DevelopmentPipeline':
525
+ """Initialize the development pipeline."""
526
+ return self.__class__.DevelopmentPipeline(
527
+ workspace_manager=self.workspace_manager,
528
+ tool_manager=self._setup_tool_manager()
529
+ )
530
+
531
+ def _setup_tool_manager(self):
532
+ """Setup tool manager with configuration."""
533
+ return ToolManager()
534
 
535
+ class RefinementLoop:
536
+ """Manages the iterative refinement process."""
537
+ def __init__(self, pipeline):
538
+ self.pipeline = pipeline
539
+ self.max_iterations = 10
540
+ self.quality_metrics = QualityMetrics()
541
+ self.logger = logging.getLogger(__name__)
542
+ self.current_iteration = 0
543
+ self.history = []
544
+
545
+ async def run_refinement_cycle(self, task: str) -> Dict[str, Any]:
546
+ """Run a complete refinement cycle for the given task."""
547
+ self.logger.info(f"Starting refinement cycle for task: {task}")
548
+ self.current_iteration = 0
549
+
550
+ try:
551
+ while self.current_iteration < self.max_iterations:
552
+ self.logger.info(f"Starting iteration {self.current_iteration + 1}")
553
+
554
+ # Execute pipeline stages
555
+ planning_result = await self.pipeline.execute_stage(
556
+ self.pipeline.PipelineStage.PLANNING,
557
+ {"task": task}
558
+ )
559
+
560
+ development_result = await self.pipeline.execute_stage(
561
+ self.pipeline.PipelineStage.DEVELOPMENT,
562
+ planning_result["result"]
563
+ )
564
+
565
+ testing_result = await self.pipeline.execute_stage(
566
+ self.pipeline.PipelineStage.TESTING,
567
+ development_result["result"]
568
  )
569
 
570
+ # Analyze results
571
+ quality_analysis = self._analyze_quality(testing_result["result"])
572
+
573
+ # Record iteration history
574
+ self.history.append({
575
+ "iteration": self.current_iteration,
576
+ "quality_metrics": quality_analysis,
577
+ "timestamp": datetime.now()
578
+ })
579
+
580
+ # Check if quality requirements are met
581
+ if self._meets_quality_requirements(quality_analysis):
582
+ self.logger.info("Quality requirements met. Refinement cycle complete.")
583
+ return self._prepare_final_result(quality_analysis)
584
+
585
+ self.current_iteration += 1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
586
 
587
+ return {
588
+ "status": "max_iterations_reached",
589
+ "iterations_completed": self.current_iteration,
590
+ "final_quality": quality_analysis
591
+ }
592
+
593
+ except Exception as e:
594
+ self.logger.error(f"Error in refinement cycle: {str(e)}")
595
+ return {"status": "error", "error": str(e)}
596
+
597
+ def _analyze_quality(self, result: Dict[str, Any]) -> Dict[str, float]:
598
+ """Analyze the quality metrics of the current iteration."""
599
+ return {
600
+ "code_quality": self.quality_metrics.code_quality_score,
601
+ "test_coverage": self.quality_metrics.test_coverage,
602
+ "security_score": float(self.quality_metrics.security_score)
603
+ }
604
+
605
+ def _meets_quality_requirements(self, quality_analysis: Dict[str, float]) -> bool:
606
+ """Check if the current quality metrics meet the requirements."""
607
+ thresholds = self.quality_metrics.thresholds
608
+ return (
609
+ quality_analysis["code_quality"] >= thresholds["code_quality"] and
610
+ quality_analysis["test_coverage"] >= thresholds["test_coverage"] and
611
+ quality_analysis["security_score"] >= thresholds["security"]
612
  )
613
+
614
+ def _prepare_final_result(self, quality_analysis: Dict[str, float]) -> Dict[str, Any]:
615
+ """Prepare the final result of the refinement cycle."""
616
+ return {
617
+ "status": "success",
618
+ "iterations_completed": self.current_iteration,
619
+ "final_quality": quality_analysis,
620
+ "history": self.history
621
+ }
622
+
623
+ def get_refinement_history(self) -> List[Dict[str, Any]]:
624
+ """Get the history of refinement iterations."""
625
+ return self.history
626
+
627
+ def run(self):
628
+ """Main application entry point."""
629
+ try:
630
+ logging.info("Starting Autonomous Agent Application")
631
+ self.interface.render_main_interface()
632
+ except Exception as e:
633
+ logging.error(f"Application error: {str(e)}")
634
+ st.error("An error occurred while starting the application. Please check the logs.")
635
+ raise
636
+
637
+ class AutonomousAgentApp:
638
+ """Main application class for the Autonomous Agent System."""
639
+
640
+ def __init__(self):
641
+ self.workspace_manager = WorkspaceManager(workspace_dir=os.getenv('WORKSPACE_DIR', 'workspace'))
642
+ self.pipeline = self._initialize_pipeline()
643
+ self.refinement_loop = self.RefinementLoop(pipeline=self.pipeline)
644
+ self.interface = self.StreamlitInterface(self)
645
+ self.autonomous_agent = self.AutonomousAgent()
646
+ self.tools_repository = self._initialize_tool_repository()
647
+ self.autonomous_agent.initialize_tool_repository(self.tools_repository)
648
+
649
+ def _initialize_tool_repository(self):
650
+ """Initializes the tool repository."""
651
+ tool_repository = ToolRepository()
652
+ tool_repository.add_tool(Tool(name="Code Review Tool", task="Review Code Quality"))
653
+ tool_repository.add_tool(Tool(name="Automated Testing Tool", task="Run Automated Tests"))
654
+ tool_repository.add_tool(Tool(name="Security Audit Tool", task="Run Security Audits"))
655
+ return tool_repository
656
+
657
+ def _initialize_pipeline(self) -> 'DevelopmentPipeline':
658
+ """Initialize the development pipeline."""
659
+ return self.DevelopmentPipeline(
660
+ workspace_manager=self.workspace_manager,
661
+ tool_manager=self._setup_tool_manager()
662
  )
663
 
664
+ def _setup_tool_manager(self):
665
+ """Setup tool manager with configuration."""
666
+ return ToolManager()
667
+
668
+ class DevelopmentPipeline:
669
+ """Manages the development pipeline stages."""
670
+
671
+ class PipelineStage(Enum):
672
+ PLANNING = "planning"
673
+ DEVELOPMENT = "development"
674
+ TESTING = "testing"
675
+ REVIEW = "review"
676
+
677
+ def __init__(self, workspace_manager: WorkspaceManager, tool_manager: 'ToolManager'):
678
+ self.workspace_manager = workspace_manager
679
+ self.tool_manager = tool_manager
680
+ self.logger = logging.getLogger(__name__)
681
+
682
+ async def execute_stage(self, stage: PipelineStage, input_data: Dict) -> Dict[str, Any]:
683
+ """Execute a pipeline stage and return results."""
684
+ self.logger.info(f"Executing pipeline stage: {stage.value}")
685
+
686
+ try:
687
+ if stage == self.PipelineStage.PLANNING:
688
+ return await self._handle_planning(input_data)
689
+ elif stage == self.PipelineStage.DEVELOPMENT:
690
+ return await self._handle_development(input_data)
691
+ elif stage == self.PipelineStage.TESTING:
692
+ return await self._handle_testing(input_data)
693
+ else:
694
+ raise ValueError(f"Unknown pipeline stage: {stage}")
695
+ except Exception as e:
696
+ self.logger.error(f"Error in {stage.value} stage: {str(e)}")
697
+ return {"status": "error", "error": str(e)}
698
+
699
+ async def _handle_planning(self, input_data: Dict) -> Dict:
700
+ """Handle planning stage execution."""
701
+ # Example implementation
702
+ return {
703
+ "status": "success",
704
+ "result": {"plan": "Sample plan"},
705
+ "artifacts": ["requirements.txt"]
706
+ }
707
+
708
+ async def _handle_development(self, input_data: Dict) -> Dict:
709
+ """Handle development stage execution."""
710
+ # Example implementation
711
+ return {
712
+ "status": "success",
713
+ "result": {"code": "print('Hello World')"},
714
+ "artifacts": ["main.py"]
715
+ }
716
+
717
+ async def _handle_testing(self, input_data: Dict) -> Dict:
718
+ """Handle testing stage execution."""
719
+ # Example implementation
720
+ return {
721
+ "status": "success",
722
+ "result": {"test_results": "All tests passed"},
723
+ "artifacts": ["test_report.html"]
724
+ }
725
+
726
+ class ToolManager:
727
+ """Manages tools used by the development pipeline."""
728
+
729
+ def __init__(self):
730
+ self.tools = {}
731
+ self.logger = logging.getLogger(__name__)
732
+
733
+ def register_tool(self, name: str, tool: Any):
734
+ """Register a new tool."""
735
+ self.tools[name] = tool
736
+ self.logger.info(f"Registered tool: {name}")
737
+
738
+ def get_tool(self, name: str) -> Optional[Any]:
739
+ """Retrieve a registered tool."""
740
+ return self.tools.get(name)
741
+
742
+ @dataclass
743
+ class QualityMetrics:
744
+ """Advanced quality metrics tracking and analysis"""
745
+ metrics_analyzer: CodeMetricsAnalyzer = None
746
+ code_quality_score: float = 0.0
747
+ test_coverage: float = 0.0
748
+ security_score: str = "unknown"
749
+ performance_score: float = 0.0
750
+
751
+ def __post_init__(self):
752
+ self.metrics_analyzer = CodeMetricsAnalyzer()
753
+ self.history = []
754
+ self.thresholds = {
755
+ "code_quality": 0.85,
756
+ "test_coverage": 0.90,
757
+ "security": 0.85,
758
+ "performance": 0.80
759
+ }
760
+
761
+ class CodeAnalyzer:
762
+ def __init__(self):
763
+ self.history = []
764
+ self.code_quality_score = 0.0
765
+ self.test_coverage = 0.0
766
+ self.security_score = "0.0"
767
+
768
+ def _get_project_files(self, project_name: str) -> list:
769
+ # Dummy implementation for example purposes
770
+ return ["file1.py", "file2.py"]
771
+
772
+ def analyze_code(self, project_name: str) -> Dict[str, Any]:
773
+ """Comprehensive code analysis"""
774
+ try:
775
+ # Get all Python files in the project
776
+ project_files = self._get_project_files(project_name)
777
+
778
+ aggregated_metrics = {
779
+ "code_quality": 0.0,
780
+ "test_coverage": 0.0,
781
+ "security": 0.0,
782
+ "performance": 0.0,
783
+ "files_analyzed": len(project_files),
784
+ "detailed_metrics": []
785
+ }
786
+
787
+ for file_path in project_files:
788
+ metrics = self.metrics_analyzer.analyze_code_quality(file_path)
789
+ aggregated_metrics["detailed_metrics"].append({
790
+ "file": file_path,
791
+ "metrics": metrics
792
+ })
793
+
794
+ # Update aggregated scores
795
+ aggregated_metrics["code_quality"] += metrics["quality_score"]
796
+ aggregated_metrics["test_coverage"] += metrics["coverage_score"]
797
+ aggregated_metrics["security"] += metrics["security_score"]
798
+
799
+ # Calculate averages
800
+ if project_files:
801
+ for key in ["code_quality", "test_coverage", "security"]:
802
+ aggregated_metrics[key] /= len(project_files)
803
+
804
+ # Update instance variables
805
+ self.code_quality_score = aggregated_metrics["code_quality"]
806
+ self.test_coverage = aggregated_metrics["test_coverage"]
807
+ self.security_score = str(aggregated_metrics["security"])
808
+
809
+ # Add to history
810
+ self.history.append({
811
+ "timestamp": datetime.now(),
812
+ "metrics": aggregated_metrics
813
+ })
814
+
815
+ return aggregated_metrics
816
+
817
+ except Exception as e:
818
+ print(f"An error occurred: {e}")
819
+ return {}
820
+
821
+ except Exception as e:
822
+ logging.error(f"Error in code analysis: {str(e)}")
823
+ return {
824
+ "error": str(e),
825
+ "code_quality": 0.0,
826
+ "test_coverage": 0.0,
827
+ "security": "error",
828
+ "performance": 0.0
829
+ }
830
+
831
+ def _get_project_files(self, project_name: str) -> List[str]:
832
+ """Get all Python files in the project"""
833
+ project_dir = os.path.join(os.getcwd(), project_name)
834
+ python_files = []
835
+
836
+ for root, _, files in os.walk(project_dir):
837
+ for file in files:
838
+ if file.endswith('.py'):
839
+ python_files.append(os.path.join(root, file))
840
+
841
+ return python_files
842
+
843
+ def get_improvement_suggestions(self) -> List[str]:
844
+ """Generate improvement suggestions based on metrics"""
845
+ suggestions = []
846
+ latest_metrics = self.history[-1]["metrics"] if self.history else None
847
+
848
+ if not latest_metrics:
849
+ return ["No metrics available for analysis"]
850
+
851
+ if latest_metrics["code_quality"] < self.thresholds["code_quality"]:
852
+ suggestions.append(
853
+ f"Code quality score ({latest_metrics['code_quality']:.2f}) is below threshold "
854
+ f"({self.thresholds['code_quality']}). Consider refactoring complex methods."
855
+ )
856
+
857
+ if latest_metrics["test_coverage"] < self.thresholds["test_ coverage"]:
858
+ suggestions.append(
859
+ f"Test coverage ({latest_metrics['test_coverage']:.2f}) is below threshold "
860
+ f"({self.thresholds['test_coverage']}). Add more unit tests."
861
+ )
862
+
863
+ if float(latest_metrics["security"]) < self.thresholds["security"]:
864
+ suggestions.append(
865
+ f"Security score ({latest_metrics['security']}) is below threshold "
866
+ f"({self.thresholds['security']}). Address security vulnerabilities."
867
+ )
868
+
869
+ return suggestions
870
+
871
+ class StreamlitInterface:
872
+ """Streamlit UI integration for the Autonomous Agent system."""
873
+
874
+ def main():
875
+ autonomous_agent_app = AutonomousAgentApp()
876
+ app.run()
877
+
878
+ def __init__(self, app: 'AutonomousAgentApp'): # Use string forward reference
879
+ self.app = app
880
+ self.chat_system = self.app.ChatSystem(self.app.autonomous_agent)
881
+
882
+ def render_main_interface(self):
883
+ """Render the main Streamlit interface."""
884
+ st.title("Autonomous Agent System")
885
+
886
+ # Add chat interface to sidebar
887
+ self.chat_system.render_chat_interface()
888
+
889
+ # Main content tabs
890
+ tab_names = ["Autonomous Agent", "Workspace Management", "Settings"]
891
+ selected_tab = st.selectbox("Select a Tab", tab_names)
892
+
893
+ if selected_tab == "Autonomous Agent":
894
+ self.render_autonomous_agent_tab()
895
+ elif selected_tab == "Workspace Management":
896
+ self.render_workspace_management_tab()
897
+ elif selected_tab == "Settings":
898
+ self.render_settings_tab()
899
+
900
+ def render_autonomous_agent_tab(self):
901
+ """Render the Autonomous Agent tab."""
902
+ st.header("Autonomous Agent")
903
+ task = st.text_area("Enter a task for the autonomous agent:")
904
+
905
+ if st.button("Run Autonomous Agent"):
906
+ if task:
907
+ # Run the autonomous agent with the provided task
908
+ try:
909
+ result = asyncio.run(self.app.refinement_loop.run_refinement_cycle(task))
910
+ st.success(f"Result: {result}")
911
+ except Exception as e:
912
+ st.error(f"An error occurred: {str(e)}")
913
+
914
+ def render_workspace_management_tab(self):
915
+ """Render the Workspace Management tab with a workspace explorer."""
916
+ st.header("Workspace Management")
917
+
918
+ # Workspace Explorer
919
+ st.subheader("Workspace Explorer")
920
+ workspace_tree = self.app.workspace_manager.get_workspace_tree()
921
+ self._render_tree(workspace_tree)
922
+
923
+ # File creation
924
+ st.subheader("Create a File")
925
+ new_filename = st.text_input("Enter filename:")
926
+ new_file_content = st.text_area("Enter file content:")
927
+ if st.button("Create File"):
928
+ if new_filename and new_file_content:
929
+ result = self.app.workspace_manager.create_file(new_filename, new_file_content)
930
+ st.success(result)
931
+ else:
932
+ st.error("Filename and content are required.")
933
+
934
+ # File deletion
935
+ st.subheader("Delete a File")
936
+ delete_filename = st.text_input("Enter filename to delete:")
937
+ if st.button("Delete File"):
938
+ if delete_filename:
939
+ result = self.app.workspace_manager.delete_file(delete_filename)
940
+ st.success(result)
941
+ else:
942
+ st.error("Filename is required.")
943
+
944
+ def _render_tree(self, tree: Dict[str, Any], level: int = 0):
945
+ """Recursively render the workspace directory tree."""
946
+ if tree["type"] == "file":
947
+ st.write(" " * level + f"📄 {tree['name']}")
948
+ elif tree["type"] == "directory":
949
+ st.write(" " * level + f"📁 {tree['name']}")
950
+ for child in tree["children"]:
951
+ self._render_tree(child, level + 1)
952
+
953
+ def render_settings_tab(self):
954
+ """Render the Settings tab."""
955
+ st.header("Application Settings")
956
+
957
+ # Section 1: Refinement Process Configuration
958
+ st.subheader("Refinement Process Settings")
959
+
960
+ # Adjust maximum refinement iterations
961
+ current_max_iter = self.app.refinement_loop.max_iterations
962
+ new_max_iter = st.number_input(
963
+ "Maximum Refinement Iterations",
964
+ min_value=1,
965
+ max_value=20,
966
+ value=current_max_iter,
967
+ help="Maximum number of refinement cycles to perform"
968
+ )
969
+ if new_max_iter != current_max_iter:
970
+ self.app.refinement_loop.max_iterations = new_max_iter
971
+ st.success(f"Updated maximum iterations to {new_max_iter}")
972
+
973
+ # Section 2: Quality Threshold Configuration
974
+ st.subheader("Quality Thresholds")
975
+
976
+ # Get current thresholds
977
+ thresholds = self.app.refinement_loop.quality_metrics.thresholds
978
+
979
+ col1, col2, col3 = st.columns(3)
980
+ with col1:
981
+ new_code_quality = st.slider(
982
+ "Code Quality Threshold",
983
+ 0.0, 1.0, thresholds["code_quality"],
984
+ help="Minimum acceptable code quality score"
985
+ )
986
+ with col2:
987
+ new_test_coverage = st.slider(
988
+ "Test Coverage Threshold",
989
+ 0.0, 1.0, thresholds["test_coverage"],
990
+ help="Minimum required test coverage"
991
+ )
992
+ with col3:
993
+ new_security = st.slider(
994
+ "Security Threshold",
995
+ 0.0, 1.0, thresholds["security"],
996
+ help="Minimum acceptable security score"
997
+ )
998
+
999
+ if st.button("Update Quality Thresholds"):
1000
+ self.app.refinement_loop.quality_metrics.thresholds.update({
1001
+ "code_quality": new_code_quality,
1002
+ "test_coverage": new_test_coverage,
1003
+ "security": new_security
1004
+ })
1005
+ st.success("Quality thresholds updated!")
1006
+
1007
+ # Section 3: Performance Configuration
1008
+ st.subheader("Performance Settings")
1009
+
1010
+ # Concurrency settings
1011
+ concurrency_level = st.selectbox(
1012
+ "Max Concurrency",
1013
+ options=[1, 2, 4, 8],
1014
+ index=2,
1015
+ help="Maximum parallel tasks for code analysis"
1016
+ )
1017
+
1018
+ # Resource limits
1019
+ mem_limit = st.slider(
1020
+ "Memory Limit (GB)",
1021
+ 1, 16, 4,
1022
+ help="Maximum memory allocation for pipeline operations"
1023
+ )
1024
+
1025
+ # Section 4: Security Settings
1026
+ st.subheader("Security Configuration")
1027
+
1028
+ # Security rules toggle
1029
+ enable_security_scan = st.checkbox(
1030
+ "Enable Real-time Security Scanning",
1031
+ value=True,
1032
+ help="Perform continuous security analysis during development"
1033
+ )
1034
+
1035
+ # Severity level filtering
1036
+ security_level = st.selectbox(
1037
+ "Minimum Security Severity Level",
1038
+ ["Low", "Medium", "High", "Critical"],
1039
+ index=1,
1040
+ help="Minimum severity level to trigger security alerts"
1041
+ )
1042
+
1043
+ # Section 5: Workspace Configuration
1044
+ st.subheader("Workspace Settings")
1045
+ current_workspace = self.app.workspace_manager.workspace_dir
1046
+ st.write(f"Current Workspace: `{current_workspace}`")
1047
+
1048
+ # Workspace actions
1049
+ if st.button("Clear Workspace Cache"):
1050
+ self.app.workspace_manager.clean_cache()
1051
+ st.success("Workspace cache cleared!")
1052
+
1053
+ # Section 6: Diagnostic Settings
1054
+ st.subheader("Diagnostics")
1055
+
1056
+ # Logging controls
1057
+ log_level = st.selectbox(
1058
+ "Logging Level",
1059
+ ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
1060
+ index=1
1061
+ )
1062
+ st.session_state.log_level = log_level # Store in session state
1063
+ logging.getLogger().setLevel(log_level)
1064
+
1065
+ # Debug mode toggle
1066
+ debug_mode = st.checkbox("Enable Debug Mode")
1067
+ st.session_state.debug_mode = debug_mode # Store in session state
1068
+ if debug_mode:
1069
+ self.app.refinement_loop.logger.setLevel(logging.DEBUG)
1070
+ else:
1071
+ self.app.refinement_loop.logger.setLevel(logging.INFO)
1072
+
1073
+ # Section 7: System Information
1074
+ st.subheader("System Info")
1075
+ st.write(f"Python Version: {sys.version}")
1076
+ st.write(f"Platform: {platform.platform()}")
1077
+ st.write(f"Available Memory: {psutil.virtual_memory().available / (1024**3):.1f} GB free")
1078
+
1079
  if __name__ == "__main__":
1080
+ app = AutonomousAgentApp() # Create an instance of the app
1081
+ app.run() # Call the run method to start the application