| import streamlit as st | |
| import google.generativeai as genai | |
| import requests | |
| import subprocess | |
| import os | |
| import pylint | |
| import pandas as pd | |
| from sklearn.model_selection import train_test_split | |
| from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier | |
| from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score | |
| import git | |
| import spacy | |
| from spacy.lang.en import English | |
| import boto3 | |
| import unittest | |
| import docker | |
| import sympy as sp | |
| from scipy.optimize import minimize, differential_evolution | |
| import numpy as np | |
| import matplotlib.pyplot as plt | |
| import seaborn as sns | |
| from IPython.display import display | |
| from tenacity import retry, stop_after_attempt, wait_fixed | |
| import torch | |
| import torch.nn as nn | |
| import torch.optim as optim | |
| from transformers import AutoTokenizer, AutoModel | |
| import networkx as nx | |
| from sklearn.cluster import KMeans | |
| from scipy.stats import ttest_ind | |
| from statsmodels.tsa.arima.model import ARIMA | |
| import nltk | |
| from nltk.sentiment import SentimentIntensityAnalyzer | |
| import cv2 | |
| from PIL import Image | |
| import tensorflow as tf | |
| from tensorflow.keras.applications import ResNet50 | |
| from tensorflow.keras.preprocessing import image | |
| from tensorflow.keras.applications.resnet50 import preprocess_input, decode_predictions | |
| # Configure the Gemini API | |
| genai.configure(api_key=st.secrets["GOOGLE_API_KEY"]) | |
| # Create the model with optimized parameters and enhanced system instructions | |
| generation_config = { | |
| "temperature": 0.4, | |
| "top_p": 0.8, | |
| "top_k": 50, | |
| "max_output_tokens": 4096, | |
| } | |
| model = genai.GenerativeModel( | |
| model_name="gemini-1.5-pro", | |
| generation_config=generation_config, | |
| system_instruction=""" | |
| You are Ath, an ultra-advanced AI code assistant with expertise across multiple domains including machine learning, data science, web development, cloud computing, and more. Your responses should showcase cutting-edge techniques, best practices, and innovative solutions. | |
| """ | |
| ) | |
| chat_session = model.start_chat(history=[]) | |
| @retry(stop=stop_after_attempt(5), wait=wait_fixed(2)) | |
| def generate_response(user_input): | |
| try: | |
| response = chat_session.send_message(user_input) | |
| return response.text | |
| except Exception as e: | |
| return f"Error: {e}" | |
| def optimize_code(code): | |
| with open("temp_code.py", "w") as file: | |
| file.write(code) | |
| result = subprocess.run(["pylint", "temp_code.py"], capture_output=True, text=True) | |
| os.remove("temp_code.py") | |
| return code | |
| def fetch_from_github(query): | |
| # Implement GitHub API interaction here | |
| pass | |
| def interact_with_api(api_url): | |
| response = requests.get(api_url) | |
| return response.json() | |
| def train_advanced_ml_model(X, y): | |
| X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2) | |
| models = { | |
| 'Random Forest': RandomForestClassifier(n_estimators=100, random_state=42), | |
| 'Gradient Boosting': GradientBoostingClassifier(n_estimators=100, random_state=42) | |
| } | |
| results = {} | |
| for name, model in models.items(): | |
| model.fit(X_train, y_train) | |
| y_pred = model.predict(X_test) | |
| results[name] = { | |
| 'accuracy': accuracy_score(y_test, y_pred), | |
| 'precision': precision_score(y_test, y_pred, average='weighted'), | |
| 'recall': recall_score(y_test, y_pred, average='weighted'), | |
| 'f1': f1_score(y_test, y_pred, average='weighted') | |
| } | |
| return results | |
| def handle_error(error): | |
| st.error(f"An error occurred: {error}") | |
| # Implement advanced error logging and notification system here | |
| def initialize_git_repo(repo_path): | |
| if not os.path.exists(repo_path): | |
| os.makedirs(repo_path) | |
| if not os.path.exists(os.path.join(repo_path, '.git')): | |
| repo = git.Repo.init(repo_path) | |
| else: | |
| repo = git.Repo(repo_path) | |
| return repo | |
| def integrate_with_git(repo_path, code): | |
| repo = initialize_git_repo(repo_path) | |
| with open(os.path.join(repo_path, "generated_code.py"), "w") as file: | |
| file.write(code) | |
| repo.index.add(["generated_code.py"]) | |
| repo.index.commit("Added generated code") | |
| def process_user_input(user_input): | |
| nlp = spacy.load("en_core_web_sm") | |
| doc = nlp(user_input) | |
| return doc | |
| def interact_with_cloud_services(service_name, action, params): | |
| client = boto3.client(service_name) | |
| response = getattr(client, action)(**params) | |
| return response | |
| def run_tests(): | |
| tests_dir = os.path.join(os.getcwd(), 'tests') | |
| if not os.path.exists(tests_dir): | |
| os.makedirs(tests_dir) | |
| init_file = os.path.join(tests_dir, '__init__.py') | |
| if not os.path.exists(init_file): | |
| with open(init_file, 'w') as f: | |
| f.write('') | |
| test_suite = unittest.TestLoader().discover(tests_dir) | |
| test_runner = unittest.TextTestRunner() | |
| test_result = test_runner.run(test_suite) | |
| return test_result | |
| def execute_code_in_docker(code): | |
| client = docker.from_env() | |
| try: | |
| container = client.containers.run( | |
| image="python:3.9", | |
| command=f"python -c '{code}'", | |
| detach=True, | |
| remove=True | |
| ) | |
| result = container.wait() | |
| logs = container.logs().decode('utf-8') | |
| return logs, result['StatusCode'] | |
| except Exception as e: | |
| return f"Error: {e}", 1 | |
| def solve_complex_equation(equation): | |
| x, y, z = sp.symbols('x y z') | |
| eq = sp.Eq(eval(equation)) | |
| solution = sp.solve(eq) | |
| return solution | |
| def advanced_optimization(function, bounds): | |
| result = differential_evolution(lambda x: eval(function), bounds) | |
| return result.x, result.fun | |
| def visualize_complex_data(data): | |
| df = pd.DataFrame(data) | |
| fig, axs = plt.subplots(2, 2, figsize=(16, 12)) | |
| sns.heatmap(df.corr(), annot=True, cmap='coolwarm', ax=axs[0, 0]) | |
| axs[0, 0].set_title('Correlation Heatmap') | |
| sns.pairplot(df, diag_kind='kde', ax=axs[0, 1]) | |
| axs[0, 1].set_title('Pairplot') | |
| df.plot(kind='box', ax=axs[1, 0]) | |
| axs[1, 0].set_title('Box Plot') | |
| sns.violinplot(data=df, ax=axs[1, 1]) | |
| axs[1, 1].set_title('Violin Plot') | |
| plt.tight_layout() | |
| return fig | |
| def analyze_complex_data(data): | |
| df = pd.DataFrame(data) | |
| summary = df.describe() | |
| correlation = df.corr() | |
| skewness = df.skew() | |
| kurtosis = df.kurtosis() | |
| return { | |
| 'summary': summary, | |
| 'correlation': correlation, | |
| 'skewness': skewness, | |
| 'kurtosis': kurtosis | |
| } | |
| def train_deep_learning_model(X, y): | |
| class DeepNN(nn.Module): | |
| def __init__(self, input_size): | |
| super(DeepNN, self).__init__() | |
| self.fc1 = nn.Linear(input_size, 64) | |
| self.fc2 = nn.Linear(64, 32) | |
| self.fc3 = nn.Linear(32, 1) | |
| def forward(self, x): | |
| x = torch.relu(self.fc1(x)) | |
| x = torch.relu(self.fc2(x)) | |
| x = torch.sigmoid(self.fc3(x)) | |
| return x | |
| X_tensor = torch.FloatTensor(X.values) | |
| y_tensor = torch.FloatTensor(y.values) | |
| model = DeepNN(X.shape[1]) | |
| criterion = nn.BCELoss() | |
| optimizer = optim.Adam(model.parameters()) | |
| epochs = 100 | |
| for epoch in range(epochs): | |
| optimizer.zero_grad() | |
| outputs = model(X_tensor) | |
| loss = criterion(outputs, y_tensor.unsqueeze(1)) | |
| loss.backward() | |
| optimizer.step() | |
| return model | |
| def perform_nlp_analysis(text): | |
| nlp = spacy.load("en_core_web_sm") | |
| doc = nlp(text) | |
| entities = [(ent.text, ent.label_) for ent in doc.ents] | |
| tokens = [token.text for token in doc] | |
| pos_tags = [(token.text, token.pos_) for token in doc] | |
| sia = SentimentIntensityAnalyzer() | |
| sentiment = sia.polarity_scores(text) | |
| return { | |
| 'entities': entities, | |
| 'tokens': tokens, | |
| 'pos_tags': pos_tags, | |
| 'sentiment': sentiment | |
| } | |
| def perform_image_analysis(image_path): | |
| img = cv2.imread(image_path) | |
| img_rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) | |
| # Perform object detection | |
| model = ResNet50(weights='imagenet') | |
| img_resized = cv2.resize(img_rgb, (224, 224)) | |
| img_array = image.img_to_array(img_resized) | |
| img_array = np.expand_dims(img_array, axis=0) | |
| img_array = preprocess_input(img_array) | |
| predictions = model.predict(img_array) | |
| decoded_predictions = decode_predictions(predictions, top=3)[0] | |
| # Perform edge detection | |
| edges = cv2.Canny(img, 100, 200) | |
| return { | |
| 'predictions': decoded_predictions, | |
| 'edges': edges | |
| } | |
| def perform_time_series_analysis(data): | |
| df = pd.DataFrame(data) | |
| model = ARIMA(df, order=(1, 1, 1)) | |
| results = model.fit() | |
| forecast = results.forecast(steps=5) | |
| return { | |
| 'model_summary': results.summary(), | |
| 'forecast': forecast | |
| } | |
| def perform_graph_analysis(nodes, edges): | |
| G = nx.Graph() | |
| G.add_nodes_from(nodes) | |
| G.add_edges_from(edges) | |
| centrality = nx.degree_centrality(G) | |
| clustering = nx.clustering(G) | |
| shortest_paths = dict(nx.all_pairs_shortest_path_length(G)) | |
| return { | |
| 'centrality': centrality, | |
| 'clustering': clustering, | |
| 'shortest_paths': shortest_paths | |
| } | |
| # Streamlit UI setup | |
| st.set_page_config(page_title="Ultra AI Code Assistant", page_icon="🚀", layout="wide") | |
| # ... (Keep the existing CSS styles) | |
| st.markdown('<div class="main-container">', unsafe_allow_html=True) | |
| st.title("🚀 Ultra AI Code Assistant") | |
| st.markdown('<p class="subtitle">Powered by Advanced AI and Domain Expertise</p>', unsafe_allow_html=True) | |
| task_type = st.selectbox("Select Task Type", [ | |
| "Code Generation", | |
| "Machine Learning", | |
| "Data Analysis", | |
| "Natural Language Processing", | |
| "Image Analysis", | |
| "Time Series Analysis", | |
| "Graph Analysis" | |
| ]) | |
| prompt = st.text_area("Enter your task description or code:", height=120) | |
| if st.button("Execute Task"): | |
| if prompt.strip() == "": | |
| st.error("Please enter a valid prompt.") | |
| else: | |
| with st.spinner("Processing your request..."): | |
| try: | |
| if task_type == "Code Generation": | |
| processed_input = process_user_input(prompt) | |
| completed_text = generate_response(processed_input.text) | |
| if "Error" in completed_text: | |
| handle_error(completed_text) | |
| else: | |
| optimized_code = optimize_code(completed_text) | |
| st.success("Code generated and optimized successfully!") | |
| st.markdown('<div class="output-container">', unsafe_allow_html=True) | |
| st.markdown('<div class="code-block">', unsafe_allow_html=True) | |
| st.code(optimized_code) | |
| st.markdown('</div>', unsafe_allow_html=True) | |
| st.markdown('</div>', unsafe_allow_html=True) | |
| repo_path = "./repo" | |
| integrate_with_git(repo_path, optimized_code) | |
| test_result = run_tests() | |
| if test_result.wasSuccessful(): | |
| st.success("All tests passed successfully!") | |
| else: | |
| st.error("Some tests failed. Please check the code.") | |
| execution_result, status_code = execute_code_in_docker(optimized_code) | |
| if status_code == 0: | |
| st.success("Code executed successfully in Docker!") | |
| st.text(execution_result) | |
| else: | |
| st.error(f"Code execution failed: {execution_result}") | |
| elif task_type == "Machine Learning": | |
| # For demonstration, we'll use a sample dataset | |
| from sklearn.datasets import make_classification | |
| X, y = make_classification(n_samples=1000, n_features=20, n_classes=2, random_state=42) | |
| results = train_advanced_ml_model(X, y) | |
| st.write("Machine Learning Model Performance:") | |
| st.json(results) | |
| st.write("Deep Learning Model:") | |
| deep_model = train_deep_learning_model(pd.DataFrame(X), pd.Series(y)) | |
| st.write(deep_model) | |
| elif task_type == "Data Analysis": | |
| # For demonstration, we'll use a sample dataset | |
| data = pd.DataFrame(np.random.randn(100, 5), columns=['A', 'B', 'C', 'D', 'E']) | |
| analysis_results = analyze_complex_data(data) | |
| st.write("Data Analysis Results:") | |
| st.write(analysis_results['summary']) | |
| st.write("Correlation Matrix:") | |
| st.write(analysis_results['correlation']) | |
| fig = visualize_complex_data(data) | |
| st.pyplot(fig) | |
| elif task_type == "Natural Language Processing": | |
| nlp_results = perform_nlp_analysis(prompt) | |
| st.write("NLP Analysis Results:") | |
| st.json(nlp_results) | |
| elif task_type == "Image Analysis": | |
| uploaded_file = st.file_uploader("Choose an image...", type=["jpg", "png", "jpeg"]) | |
| if uploaded_file is not None: | |
| image = Image.open(uploaded_file) | |
| st.image(image, caption='Uploaded Image', use_column_width=True) | |
| # Save the uploaded image temporarily | |
| with open("temp_image.jpg", "wb") as f: | |
| f.write(uploaded_file.getbuffer()) | |
| analysis_results = perform_image_analysis("temp_image.jpg") | |
| st.write("Image Analysis Results:") | |
| st.write("Top 3 predictions:") | |
| for i, (imagenet_id, label, score) in enumerate(analysis_results['predictions']): | |
| st.write(f"{i + 1}: {label} ({score:.2f})") | |
| st.write("Edge Detection:") | |
| st.image(analysis_results['edges'], caption='Edge Detection', use_column_width=True) | |
| # Remove the temporary image file | |
| os.remove("temp_image.jpg") | |
| elif task_type == "Time Series Analysis": | |
| # For demonstration, we'll use a sample time series dataset | |
| dates = pd.date_range(start='1/1/2020', end='1/1/2021', freq='D') | |
| values = np.random.randn(len(dates)).cumsum() | |
| ts_data = pd.Series(values, index=dates) | |
| st.line_chart(ts_data) | |
| analysis_results = perform_time_series_analysis(ts_data) | |
| st.write("Time Series Analysis Results:") | |
| st.write(analysis_results['model_summary']) | |
| st.write("Forecast for the next 5 periods:") | |
| st.write(analysis_results['forecast']) | |
| elif task_type == "Graph Analysis": | |
| # For demonstration, we'll use a sample graph | |
| nodes = range(1, 11) | |
| edges = [(1, 2), (1, 3), (2, 4), (2, 5), (3, 6), (3, 7), (4, 8), (5, 9), (6, 10)] | |
| analysis_results = perform_graph_analysis(nodes, edges) | |
| st.write("Graph Analysis Results:") | |
| st.write("Centrality:") | |
| st.json(analysis_results['centrality']) | |
| st.write("Clustering Coefficient:") | |
| st.json(analysis_results['clustering']) | |
| # Visualize the graph | |
| G = nx.Graph() | |
| G.add_nodes_from(nodes) | |
| G.add_edges_from(edges) | |
| fig, ax = plt.subplots(figsize=(10, 8)) | |
| nx.draw(G, with_labels=True, node_color='lightblue', node_size=500, font_size=16, font_weight='bold', ax=ax) | |
| st.pyplot(fig) | |
| except Exception as e: | |
| handle_error(e) | |
| st.markdown(""" | |
| <div style='text-align: center; margin-top: 2rem; color: #4a5568;'> | |
| Created with ❤️ by Your Ultra AI Code Assistant | |
| </div> | |
| """, unsafe_allow_html=True) | |
| st.markdown('</div>', unsafe_allow_html=True) | |
| # Additional helper functions | |
| def explain_code(code): | |
| """Generate an explanation for the given code using NLP techniques.""" | |
| explanation = generate_response(f"Explain the following code:\n\n{code}") | |
| return explanation | |
| def generate_unit_tests(code): | |
| """Generate unit tests for the given code.""" | |
| unit_tests = generate_response(f"Generate unit tests for the following code:\n\n{code}") | |
| return unit_tests | |
| def suggest_optimizations(code): | |
| """Suggest optimizations for the given code.""" | |
| optimizations = generate_response(f"Suggest optimizations for the following code:\n\n{code}") | |
| return optimizations | |
| def generate_documentation(code): | |
| """Generate documentation for the given code.""" | |
| documentation = generate_response(f"Generate documentation for the following code:\n\n{code}") | |
| return documentation | |
| # Add these new functions to the Streamlit UI | |
| if task_type == "Code Generation": | |
| st.sidebar.header("Code Analysis Tools") | |
| if st.sidebar.button("Explain Code"): | |
| explanation = explain_code(optimized_code) | |
| st.sidebar.subheader("Code Explanation") | |
| st.sidebar.write(explanation) | |
| if st.sidebar.button("Generate Unit Tests"): | |
| unit_tests = generate_unit_tests(optimized_code) | |
| st.sidebar.subheader("Generated Unit Tests") | |
| st.sidebar.code(unit_tests) | |
| if st.sidebar.button("Suggest Optimizations"): | |
| optimizations = suggest_optimizations(optimized_code) | |
| st.sidebar.subheader("Suggested Optimizations") | |
| st.sidebar.write(optimizations) | |
| if st.sidebar.button("Generate Documentation"): | |
| documentation = generate_documentation(optimized_code) | |
| st.sidebar.subheader("Generated Documentation") | |
| st.sidebar.write(documentation) | |
| # Add more advanced features | |
| def perform_security_analysis(code): | |
| """Perform a basic security analysis on the given code.""" | |
| security_analysis = generate_response(f"Perform a security analysis on the following code and suggest improvements:\n\n{code}") | |
| return security_analysis | |
| def generate_api_documentation(code): | |
| """Generate API documentation for the given code.""" | |
| api_docs = generate_response(f"Generate API documentation for the following code:\n\n{code}") | |
| return api_docs | |
| def suggest_design_patterns(code): | |
| """Suggest appropriate design patterns for the given code.""" | |
| design_patterns = generate_response(f"Suggest appropriate design patterns for the following code:\n\n{code}") | |
| return design_patterns | |
| # Add these new functions to the Streamlit UI | |
| if task_type == "Code Generation": | |
| st.sidebar.header("Advanced Code Analysis") | |
| if st.sidebar.button("Security Analysis"): | |
| security_analysis = perform_security_analysis(optimized_code) | |
| st.sidebar.subheader("Security Analysis") | |
| st.sidebar.write(security_analysis) | |
| if st.sidebar.button("Generate API Documentation"): | |
| api_docs = generate_api_documentation(optimized_code) | |
| st.sidebar.subheader("API Documentation") | |
| st.sidebar.write(api_docs) | |
| if st.sidebar.button("Suggest Design Patterns"): | |
| design_patterns = suggest_design_patterns(optimized_code) | |
| st.sidebar.subheader("Suggested Design Patterns") | |
| st.sidebar.write(design_patterns) | |
| # Add a feature to generate a complete project structure | |
| def generate_project_structure(project_description): | |
| """Generate a complete project structure based on the given description.""" | |
| project_structure = generate_response(f"Generate a complete project structure for the following project description:\n\n{project_description}") | |
| return project_structure | |
| # Add this new function to the Streamlit UI | |
| if st.sidebar.button("Generate Project Structure"): | |
| project_description = st.sidebar.text_area("Enter project description:") | |
| if project_description: | |
| project_structure = generate_project_structure(project_description) | |
| st.sidebar.subheader("Generated Project Structure") | |
| st.sidebar.code(project_structure) | |
| # Add a feature to suggest relevant libraries and frameworks | |
| def suggest_libraries(code): | |
| """Suggest relevant libraries and frameworks for the given code.""" | |
| suggestions = generate_response(f"Suggest relevant libraries and frameworks for the following code:\n\n{code}") | |
| return suggestions | |
| # Add this new function to the Streamlit UI | |
| if task_type == "Code Generation": | |
| if st.sidebar.button("Suggest Libraries"): | |
| library_suggestions = suggest_libraries(optimized_code) | |
| st.sidebar.subheader("Suggested Libraries and Frameworks") | |
| st.sidebar.write(library_suggestions) | |
| # Add a feature to generate code in multiple programming languages | |
| def translate_code(code, target_language): | |
| """Translate the given code to the specified target language.""" | |
| translated_code = generate_response(f"Translate the following code to {target_language}:\n\n{code}") | |
| return translated_code | |
| # Add this new function to the Streamlit UI | |
| if task_type == "Code Generation": | |
| target_language = st.sidebar.selectbox("Select target language for translation", ["Python", "JavaScript", "Java", "C++", "Go"]) | |
| if st.sidebar.button("Translate Code"): | |
| translated_code = translate_code(optimized_code, target_language) | |
| st.sidebar.subheader(f"Translated Code ({target_language})") | |
| st.sidebar.code(translated_code) | |
| # Add a feature to generate a README file for the project | |
| def generate_readme(project_description, code): | |
| """Generate a README file for the project based on the description and code.""" | |
| readme_content = generate_response(f"Generate a README.md file for the following project:\n\nDescription: {project_description}\n\nCode:\n{code}") | |
| return readme_content | |
| # Add this new function to the Streamlit UI | |
| if task_type == "Code Generation": | |
| if st.sidebar.button("Generate README"): | |
| project_description = st.sidebar.text_area("Enter project description:") | |
| if project_description: | |
| readme_content = generate_readme(project_description, optimized_code) | |
| st.sidebar.subheader("Generated README.md") | |
| st.sidebar.markdown(readme_content) | |
| # Add a feature to suggest code refactoring | |
| def suggest_refactoring(code): | |
| """Suggest code refactoring improvements for the given code.""" | |
| refactoring_suggestions = generate_response(f"Suggest code refactoring improvements for the following code:\n\n{code}") | |
| return refactoring_suggestions | |
| # Add this new function to the Streamlit UI | |
| if task_type == "Code Generation": | |
| if st.sidebar.button("Suggest Refactoring"): | |
| refactoring_suggestions = suggest_refactoring(optimized_code) | |
| st.sidebar.subheader("Refactoring Suggestions") | |
| st.sidebar.write(refactoring_suggestions) | |
| # Add a feature to generate sample test data | |
| def generate_test_data(code): | |
| """Generate sample test data for the given code.""" | |
| test_data = generate_response(f"Generate sample test data for the following code:\n\n{code}") | |
| return test_data | |
| # Add this new function to the Streamlit UI | |
| if task_type == "Code Generation": | |
| if st.sidebar.button("Generate Test Data"): | |
| test_data = generate_test_data(optimized_code) | |
| st.sidebar.subheader("Generated Test Data") | |
| st.sidebar.code(test_data) | |
| # Main execution | |
| if __name__ == "__main__": | |
| st.sidebar.header("About") | |
| st.sidebar.info("This Ultra AI Code Assistant is powered by advanced AI models and incorporates expertise across multiple domains including software development, machine learning, data analysis, and more.") | |
| st.sidebar.header("Feedback") | |
| feedback = st.sidebar.text_area("Please provide any feedback or suggestions:") | |
| if st.sidebar.button("Submit Feedback"): | |
| # Here you would typically send this feedback to a database or email | |
| st.sidebar.success("Thank you for your feedback!") |