import json
import logging
import yaml
from pathlib import Path
from tqdm.auto import tqdm
from argparse import ArgumentParser
try:
from .server import load_content
except ImportError:
from server import load_content
logger = logging.getLogger(__name__)
logging.getLogger("simple_parsing").setLevel(logging.INFO)
TEMPLATE = """
Trajectory Viewer
{file_path_tree}
Conversation History
{file_content}
"""
try:
with open(Path(__file__).parent / 'style.css', 'r') as infile:
STYLE_SHEET = infile.read()
except Exception as e:
style_file = Path(__file__).parent / 'style.css'
logger.error(f"Failed to load style sheet from {style_file}: {e}")
raise e
def _load_file(file_name, gold_patches, test_patches):
try:
role_map = {
"user": "Computer",
"assistant": "SWE-Agent",
"subroutine": "SWE-Agent subroutine",
"default": "Default",
"system": "System",
"demo": "Demonstration",
}
content = load_content(file_name, gold_patches, test_patches)
if 'history' in content and isinstance(content['history'], list):
history_content = ""
for index, item in enumerate(content['history']):
item_content = item.get('content', '').replace('<', '<').replace('>', '>')
if item.get('agent') and item['agent'] != "primary":
role_class = "subroutine"
else:
role_class = item.get('role', 'default').lower().replace(' ', '-')
element_id = f"historyItem{index}"
role_name = role_map.get(item.get('role', ''), item.get('role', ''))
history_content += (
f''''''
f'''
{role_name}
'''
f'''
'''
f'''
{item_content}
'''
f'''
'''
f'''
'''
f'''
'''
)
return history_content
else:
return 'No history content found.'
except Exception as e:
return f"Error loading content. {e}"
def _make_file_path_tree(file_path):
path_parts = file_path.split('/')
relevant_parts = path_parts[-3:]
html_string = '\n'
for part in relevant_parts:
html_string += f'
{part}
\n'
html_string += '
'
return html_string
def save_static_viewer(file_path):
if not isinstance(file_path, Path):
file_path = Path(file_path)
data = []
if "args.yaml" in list(map(lambda x: x.name, file_path.parent.iterdir())):
args = yaml.safe_load(open(file_path.parent / "args.yaml", "r"))
if "environment" in args and "data_path" in args["environment"]:
data_path = Path(__file__).parent.parent / args["environment"]["data_path"]
if data_path.exists():
data = json.load(open(data_path, "r"))
if (
not isinstance(data, list) or
not data or
'patch' not in data[0] or
'test_patch' not in data[0]
):
data = []
gold_patches = {x["instance_id"]: x["patch"] for x in data}
test_patches = {x["instance_id"]: x["test_patch"] for x in data}
content = _load_file(file_path, gold_patches, test_patches)
file_path_tree = _make_file_path_tree(file_path.absolute().as_posix())
icons_path = Path(__file__).parent / 'icons'
relative_icons_path = find_relative_path(file_path, icons_path)
style_sheet = STYLE_SHEET.replace(
"url('icons/",
f"url('{relative_icons_path.as_posix()}/"
).replace(
'url("icons/',
f'url("{relative_icons_path.as_posix()}/'
)
data = TEMPLATE.format(file_content=content, style_sheet=style_sheet, file_path_tree=file_path_tree)
output_file = file_path.with_suffix('.html')
with open(output_file, 'w+') as outfile:
print(data, file=outfile)
logger.info(f"Saved static viewer to {output_file}")
def find_relative_path(from_path, to_path):
# Convert paths to absolute for uniformity
from_path = from_path.resolve()
to_path = to_path.resolve()
if from_path.is_file():
from_path = from_path.parent
if to_path.is_file():
to_path = to_path.parent
if not from_path.is_dir() or not to_path.is_dir():
raise ValueError(f"Both from_path and to_path must be directories, but got {from_path} and {to_path}")
# Identify the common ancestor and the parts of each path beyond it
common_parts = 0
for from_part, to_part in zip(from_path.parts, to_path.parts):
if from_part != to_part:
break
common_parts += 1
# Calculate the '../' needed to get back from from_path to the common ancestor
back_to_ancestor = ['..'] * (len(from_path.parts) - common_parts)
# Direct path from common ancestor to to_path
to_target = to_path.parts[common_parts:]
# Combine to get the relative path
relative_path = Path(*back_to_ancestor, *to_target)
return relative_path
def save_all_trajectories(directory):
if not isinstance(directory, Path):
directory = Path(directory)
all_files = list(directory.glob('**/*.traj'))
logger.info(f"Found {len(all_files)} trajectory files in {directory}")
for file_path in tqdm(all_files, desc="Saving static viewers"):
save_static_viewer(file_path)
logger.info(f"Saved static viewers for all trajectories in {args.directory}")
if __name__ == "__main__":
parser = ArgumentParser()
parser.add_argument("directory", type=str, help="Directory containing trajectory files")
args = parser.parse_args()
save_all_trajectories(args.directory)