|
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 |
|
|
