Spaces:
Sleeping
Sleeping
| import json | |
| import gradio as gr | |
| import os | |
| import requests | |
| from huggingface_hub import AsyncInferenceClient | |
| HF_TOKEN = os.getenv('HF_TOKEN') | |
| api_url = os.getenv('API_URL') | |
| headers = {"Authorization": f"Bearer {HF_TOKEN}"} | |
| client = AsyncInferenceClient(api_url) | |
| system_prompt = """ | |
| ### Instruction: | |
| Refactor the provided Python code to improve its maintainability and efficiency and reduce complexity. Include the refactored code along with the comments on the changes made for improving the metrics. | |
| ### Input: | |
| """ | |
| title = "Python Refactoring" | |
| description = """ | |
| Please give it 4 to 5 minutes for the model to load and Run , consider using Python code with less than 120 lines of code due to GPU constrainst | |
| """ | |
| css = """.toast-wrap { display: none !important } """ | |
| examples=[ [""" | |
| import sys | |
| import os | |
| import someDatabaseLib | |
| # Global variables | |
| config = {"db": "localhost", "user": "admin", "password": "admin"} | |
| connection = None | |
| def dbConnect(): | |
| global connection | |
| try: | |
| connection = someDatabaseLib.connect(config["db"], config["user"], config["password"]) | |
| except Exception as e: | |
| print(e) | |
| sys.exit(1) | |
| def fetchData(): | |
| global connection | |
| if connection is None: | |
| print("Not connected to DB") | |
| return None | |
| try: | |
| cursor = connection.cursor() | |
| cursor.execute("SELECT * FROM someTable WHERE someColumn='someValue'") | |
| return cursor.fetchall() | |
| except Exception as e: | |
| print("Failed to fetch data: ", e) | |
| return None | |
| def processData(data): | |
| if data is None: | |
| print("No data provided") | |
| return None | |
| result = [] | |
| for row in data: | |
| # Processing logic here | |
| result.append(row) | |
| return result | |
| def main(): | |
| dbConnect() | |
| data = fetchData() | |
| if data is None: | |
| print("No data fetched") | |
| sys.exit(1) | |
| processedData = processData(data) | |
| print("Data processed") | |
| if __name__ == "__main__": | |
| main() | |
| # Additional functions and logic mixed together without clear separation or modularisation | |
| def someOtherFunction(): | |
| pass | |
| # Hardcoded paths and configuration details | |
| path_to_files = "/path/to/some/files" | |
| for file_name in os.listdir(path_to_files): | |
| with open(os.path.join(path_to_files, file_name), 'r') as file: | |
| data = file.read() | |
| # Do something with the data | |
| # Poor error handling and mixing of concerns (e.g., UI logic with business logic) | |
| def userInterfaceFunction(): | |
| choice = input("Enter your choice: ") | |
| if choice == "1": | |
| print("User chose 1") | |
| # Proceed with option 1 | |
| elif choice == "2": | |
| print("User chose 2") | |
| # Proceed with option 2 | |
| else: | |
| print("Invalid choice") | |
| # Direct database access mixed with business logic without any abstraction layer | |
| def directDBAccess(): | |
| global config | |
| try: | |
| conn = someDatabaseLib.connect(config["db"], config["user"], config["password"]) | |
| cursor = conn.cursor() | |
| cursor.execute("UPDATE someTable SET someColumn='newValue' WHERE anotherColumn='value'") | |
| except Exception as e: | |
| print("Database operation failed: ", e) | |
| # Mixing of different levels of abstraction, lack of consistent error handling, and no use of classes or functions to encapsulate related operations | |
| """] , | |
| [""" | |
| def analyze_sales_data(sales_records): | |
| active_sales = filter(lambda record: record['status'] == 'active', sales_records) | |
| sales_by_category = {} | |
| for record in active_sales: | |
| category = record['category'] | |
| total_sales = record['units_sold'] * record['price_per_unit'] | |
| if category not in sales_by_category: | |
| sales_by_category[category] = {'total_sales': 0, 'total_units': 0} | |
| sales_by_category[category]['total_sales'] += total_sales | |
| sales_by_category[category]['total_units'] += record['units_sold'] | |
| average_sales_data = [] | |
| for category, data in sales_by_category.items(): | |
| average_sales = data['total_sales'] / data['total_units'] | |
| sales_by_category[category]['average_sales'] = average_sales | |
| average_sales_data.append((category, average_sales)) | |
| average_sales_data.sort(key=lambda x: x[1], reverse=True) | |
| for rank, (category, _) in enumerate(average_sales_data, start=1): | |
| sales_by_category[category]['rank'] = rank | |
| return sales_by_category | |
| """] , | |
| [""" | |
| import pandas as pd | |
| import re | |
| import ast | |
| from code_bert_score import score | |
| import numpy as np | |
| def preprocess_code(source_text): | |
| def remove_comments_and_docstrings(source_code): | |
| source_code = re.sub(r'#.*', '', source_code) | |
| source_code = re.sub(r'(\'\'\'(.*?)\'\'\'|\"\"\"(.*?)\"\"\")', '', source_code, flags=re.DOTALL) | |
| return source_code | |
| pattern = r"```python\s+(.+?)\s+```" | |
| matches = re.findall(pattern, source_text, re.DOTALL) | |
| code_to_process = '\n'.join(matches) if matches else source_text | |
| cleaned_code = remove_comments_and_docstrings(code_to_process) | |
| return cleaned_code | |
| def evaluate_dataframe(df): | |
| results = {'P': [], 'R': [], 'F1': [], 'F3': []} | |
| for index, row in df.iterrows(): | |
| try: | |
| cands = [preprocess_code(row['generated_text'])] | |
| refs = [preprocess_code(row['output'])] | |
| P, R, F1, F3 = score(cands, refs, lang='python') | |
| results['P'].append(P[0]) | |
| results['R'].append(R[0]) | |
| results['F1'].append(F1[0]) | |
| results['F3'].append(F3[0]) | |
| except Exception as e: | |
| print(f"Error processing row {index}: {e}") | |
| for key in results.keys(): | |
| results[key].append(None) | |
| df_metrics = pd.DataFrame(results) | |
| return df_metrics | |
| def evaluate_dataframe_multiple_runs(df, runs=3): | |
| all_results = [] | |
| for run in range(runs): | |
| df_metrics = evaluate_dataframe(df) | |
| all_results.append(df_metrics) | |
| # Calculate mean and std deviation of metrics across runs | |
| df_metrics_mean = pd.concat(all_results).groupby(level=0).mean() | |
| df_metrics_std = pd.concat(all_results).groupby(level=0).std() | |
| return df_metrics_mean, df_metrics_std | |
| """ ] ] | |
| # Stream text - stream tokens with InferenceClient from TGI | |
| async def predict(message, chatbot, temperature=0.1, max_new_tokens=4096, top_p=0.6, repetition_penalty=1.15,): | |
| temperature = float(temperature) | |
| if temperature < 1e-2: | |
| temperature = 1e-2 | |
| top_p = float(top_p) | |
| input_prompt = system_prompt + str(message) + " [/INST] " | |
| partial_message = "" | |
| async for token in await client.text_generation(prompt=input_prompt, | |
| max_new_tokens=max_new_tokens, | |
| stream=True, | |
| best_of=1, | |
| temperature=temperature, | |
| top_p=top_p, | |
| do_sample=True, | |
| repetition_penalty=repetition_penalty): | |
| partial_message = partial_message + token | |
| yield partial_message | |
| gr.ChatInterface( | |
| predict, | |
| chatbot=gr.Chatbot(height=500), | |
| textbox=gr.Textbox(lines=10, label="Python Code" , placeholder="Enter or Paste your Python code here..."), | |
| title=title, | |
| description=description, | |
| theme="abidlabs/Lime", | |
| examples=examples, | |
| cache_examples=False, | |
| submit_btn = "Submit_code", | |
| retry_btn="Retry", | |
| undo_btn="Undo", | |
| clear_btn="Clear", | |
| ).queue().launch() | |