| import os |
| import re |
| import shutil |
| import signal |
| import subprocess |
| import time |
| from typing import Dict |
|
|
| import openai |
| import requests |
|
|
| from chatdev.codes import Codes |
| from chatdev.documents import Documents |
| from chatdev.roster import Roster |
| from chatdev.utils import log_and_print_online |
|
|
|
|
| class ChatEnvConfig: |
| def __init__(self, clear_structure, |
| brainstorming, |
| gui_design, |
| git_management): |
| self.clear_structure = clear_structure |
| self.brainstorming = brainstorming |
| self.gui_design = gui_design |
| self.git_management = git_management |
|
|
| def __str__(self): |
| string = "" |
| string += "ChatEnvConfig.clear_structure: {}\n".format(self.clear_structure) |
| string += "ChatEnvConfig.brainstorming: {}\n".format(self.brainstorming) |
| return string |
|
|
|
|
| class ChatEnv: |
| def __init__(self, chat_env_config: ChatEnvConfig): |
| self.config = chat_env_config |
| self.roster: Roster = Roster() |
| self.codes: Codes = Codes() |
| self.proposed_images: Dict[str, str] = {} |
| self.incorporated_images: Dict[str, str] = {} |
| self.requirements: Documents = Documents() |
| self.manuals: Documents = Documents() |
| self.env_dict = { |
| "directory": "", |
| "task_prompt": "", |
| "modality": "", |
| "ideas": "", |
| "language": "", |
| "review_comments": "", |
| "error_summary": "", |
| "test_reports": "" |
| } |
|
|
| @staticmethod |
| def fix_module_not_found_error(test_reports): |
| if "ModuleNotFoundError" in test_reports: |
| for match in re.finditer(r"No module named '(\S+)'", test_reports, re.DOTALL): |
| module = match.group(1) |
| subprocess.Popen("pip install {}".format(module), shell=True).wait() |
| log_and_print_online("**[CMD Execute]**\n\n[CMD] pip install {}".format(module)) |
|
|
| def set_directory(self, directory): |
| assert len(self.env_dict['directory']) == 0 |
| self.env_dict['directory'] = directory |
| self.codes.directory = directory |
| self.requirements.directory = directory |
| self.manuals.directory = directory |
|
|
| if os.path.exists(self.env_dict['directory']) and len(os.listdir(directory)) > 0: |
| new_directory = "{}.{}".format(directory, time.strftime("%Y%m%d%H%M%S", time.localtime())) |
| shutil.copytree(directory, new_directory) |
| print("{} Copied to {}".format(directory, new_directory)) |
| if self.config.clear_structure: |
| if os.path.exists(self.env_dict['directory']): |
| shutil.rmtree(self.env_dict['directory']) |
| os.mkdir(self.env_dict['directory']) |
| print("{} Created".format(directory)) |
| else: |
| os.mkdir(self.env_dict['directory']) |
|
|
| def exist_bugs(self) -> tuple[bool, str]: |
| directory = self.env_dict['directory'] |
|
|
| success_info = "The software run successfully without errors." |
| try: |
| command = "cd {}; ls -l; python3 main.py;".format(directory) |
| process = subprocess.Popen(command, shell=True, preexec_fn=os.setsid, |
| stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| time.sleep(3) |
| return_code = process.returncode |
| |
| if process.poll() is None: |
| os.killpg(os.getpgid(process.pid), signal.SIGTERM) |
| if return_code == 0: |
| return False, success_info |
| else: |
| error_output = process.stderr.read().decode('utf-8') |
| if error_output: |
| if "Traceback".lower() in error_output.lower(): |
| errs = error_output.replace(directory + "/", "") |
| return True, errs |
| else: |
| return False, success_info |
| except subprocess.CalledProcessError as e: |
| return True, f"Error: {e}" |
| except Exception as ex: |
| return True, f"An error occurred: {ex}" |
|
|
| return False, success_info |
|
|
| def recruit(self, agent_name: str): |
| self.roster._recruit(agent_name) |
|
|
| def exist_employee(self, agent_name: str) -> bool: |
| return self.roster._exist_employee(agent_name) |
|
|
| def print_employees(self): |
| self.roster._print_employees() |
|
|
| def update_codes(self, generated_content): |
| self.codes._update_codes(generated_content) |
|
|
| def rewrite_codes(self) -> None: |
| self.codes._rewrite_codes(self.config.git_management) |
|
|
| def get_codes(self) -> str: |
| return self.codes._get_codes() |
|
|
| def _load_from_hardware(self, directory) -> None: |
| self.codes._load_from_hardware(directory) |
|
|
| def _update_requirements(self, generated_content): |
| self.requirements._update_docs(generated_content) |
|
|
| def rewrite_requirements(self): |
| self.requirements._rewrite_docs() |
|
|
| def get_requirements(self) -> str: |
| return self.requirements._get_docs() |
|
|
| def _update_manuals(self, generated_content): |
| self.manuals._update_docs(generated_content, parse=False, predifined_filename="manual.md") |
|
|
| def rewrite_manuals(self): |
| self.manuals._rewrite_docs() |
|
|
| def write_meta(self) -> None: |
| directory = self.env_dict['directory'] |
|
|
| if not os.path.exists(directory): |
| os.mkdir(directory) |
| print("{} Created.".format(directory)) |
|
|
| meta_filename = "meta.txt" |
| with open(os.path.join(directory, meta_filename), "w", encoding="utf-8") as writer: |
| writer.write("{}:\n{}\n\n".format("Task", self.env_dict['task_prompt'])) |
| writer.write("{}:\n{}\n\n".format("Config", self.config.__str__())) |
| writer.write("{}:\n{}\n\n".format("Roster", ", ".join(self.roster.agents))) |
| writer.write("{}:\n{}\n\n".format("Modality", self.env_dict['modality'])) |
| writer.write("{}:\n{}\n\n".format("Ideas", self.env_dict['ideas'])) |
| writer.write("{}:\n{}\n\n".format("Language", self.env_dict['language'])) |
| writer.write("{}:\n{}\n\n".format("Code_Version", self.codes.version)) |
| writer.write("{}:\n{}\n\n".format("Proposed_images", len(self.proposed_images.keys()))) |
| writer.write("{}:\n{}\n\n".format("Incorporated_images", len(self.incorporated_images.keys()))) |
| print(os.path.join(directory, meta_filename), "Wrote") |
|
|
| def generate_images_from_codes(self): |
| def download(img_url, file_name): |
| r = requests.get(img_url) |
| filepath = os.path.join(self.env_dict['directory'], file_name) |
| if os.path.exists(filepath): |
| os.remove(filepath) |
| with open(filepath, "wb") as f: |
| f.write(r.content) |
| print("{} Downloaded".format(filepath)) |
|
|
| regex = r"(\w+.png)" |
| joined_codes = self.get_codes() |
| matches = re.finditer(regex, joined_codes, re.DOTALL) |
| |
| for match in matches: |
| filename = match.group(1).strip() |
| if filename in self.proposed_images.keys(): |
| self.incorporated_images[filename] = self.proposed_images[filename] |
| else: |
| self.incorporated_images[filename] = filename.replace("_", " ") |
|
|
| for filename in self.incorporated_images.keys(): |
| if not os.path.exists(os.path.join(self.env_dict['directory'], filename)): |
| desc = self.incorporated_images[filename] |
| if desc.endswith(".png"): |
| desc = desc.replace(".png", "") |
| print("{}: {}".format(filename, desc)) |
| response = openai.Image.create( |
| prompt=desc, |
| n=1, |
| size="256x256" |
| ) |
| image_url = response['data'][0]['url'] |
| download(image_url, filename) |
|
|
| def get_proposed_images_from_message(self, messages): |
| def download(img_url, file_name): |
| r = requests.get(img_url) |
| filepath = os.path.join(self.env_dict['directory'], file_name) |
| if os.path.exists(filepath): |
| os.remove(filepath) |
| with open(filepath, "wb") as f: |
| f.write(r.content) |
| print("{} Downloaded".format(filepath)) |
|
|
| regex = r"(\w+.png):(.*?)\n" |
| matches = re.finditer(regex, messages, re.DOTALL) |
| images = {} |
| for match in matches: |
| filename = match.group(1).strip() |
| desc = match.group(2).strip() |
| images[filename] = desc |
|
|
| if len(images.keys()) == 0: |
| regex = r"(\w+.png)" |
| matches = re.finditer(regex, messages, re.DOTALL) |
| images = {} |
| for match in matches: |
| filename = match.group(1).strip() |
| desc = " ".join(filename.replace(".png", "").split("_")) |
| images[filename] = desc |
| print("{}: {}".format(filename, images[filename])) |
|
|
| for filename in images.keys(): |
| if not os.path.exists(os.path.join(self.env_dict['directory'], filename)): |
| desc = images[filename] |
| if desc.endswith(".png"): |
| desc = desc.replace(".png", "") |
| print("{}: {}".format(filename, desc)) |
| response = openai.Image.create( |
| prompt=desc, |
| n=1, |
| size="256x256" |
| ) |
| image_url = response['data'][0]['url'] |
| download(image_url, filename) |
|
|
| return images |
|
|
