Update virtual_fs.py
This commit is contained in:
+23
-60
@@ -1,76 +1,39 @@
|
|||||||
"""
|
"""
|
||||||
Virtual file system for the deep agent.
|
Virtual file system used by the deep agent.
|
||||||
|
It stores files in memory and can be dumped to the real file system.
|
||||||
This module implements a simple in-memory file system that allows the agent to create
|
|
||||||
and modify files during its execution. At the end of the run the virtual files can
|
|
||||||
be dumped to the real file system.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
import os
|
||||||
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Dict, Iterable, Tuple
|
from typing import Dict, Optional
|
||||||
|
|
||||||
|
|
||||||
class VirtualFileSystem:
|
class VirtualFileSystem:
|
||||||
"""A minimal in‑memory file system.
|
"""A simple in‑memory file system.
|
||||||
|
|
||||||
The files are stored in a dictionary mapping a relative path to its content.
|
Files are stored as a mapping from a relative path to its content.
|
||||||
The content is stored as a string. The class provides methods for writing
|
|
||||||
files, reading files and dumping all files to a real directory.
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self) -> None:
|
def __init__(self, base_dir: str = "virtual_fs"):
|
||||||
self._files: Dict[str, str] = {}
|
self.base_dir = Path(base_dir)
|
||||||
|
self.files: Dict[str, str] = {}
|
||||||
|
|
||||||
def write(self, path: str | Path, content: str) -> None:
|
def write_file(self, path: str, content: str) -> None:
|
||||||
"""Write *content* to *path*.
|
norm_path = Path(path).as_posix()
|
||||||
|
self.files[norm_path] = content
|
||||||
|
|
||||||
Parameters
|
def read_file(self, path: str) -> Optional[str]:
|
||||||
----------
|
norm_path = Path(path).as_posix()
|
||||||
path:
|
return self.files.get(norm_path)
|
||||||
The relative path of the file. ``Path`` objects are accepted for
|
|
||||||
convenience.
|
|
||||||
content:
|
|
||||||
The textual content to store.
|
|
||||||
"""
|
|
||||||
path = str(path).lstrip("./")
|
|
||||||
self._files[path] = content
|
|
||||||
|
|
||||||
def read(self, path: str | Path) -> str:
|
def list_files(self) -> Dict[str, str]:
|
||||||
"""Return the content of *path*.
|
return dict(self.files)
|
||||||
|
|
||||||
Raises
|
def dump_to_real_fs(self, real_root: str = "real_fs") -> None:
|
||||||
------
|
real_root_path = Path(real_root)
|
||||||
KeyError
|
for rel_path, content in self.files.items():
|
||||||
If the file does not exist.
|
file_path = real_root_path / rel_path
|
||||||
"""
|
|
||||||
path = str(path).lstrip("./")
|
|
||||||
return self._files[path]
|
|
||||||
|
|
||||||
def list_files(self) -> Iterable[Tuple[str, str]]:
|
|
||||||
"""Yield ``(path, content)`` for all stored files."""
|
|
||||||
for path, content in self._files.items():
|
|
||||||
yield path, content
|
|
||||||
|
|
||||||
def dump_to_disk(self, root: str | Path) -> None:
|
|
||||||
"""Write all virtual files to *root*.
|
|
||||||
|
|
||||||
Parameters
|
|
||||||
----------
|
|
||||||
root:
|
|
||||||
The directory on the real file system where the files should be
|
|
||||||
created.
|
|
||||||
"""
|
|
||||||
root = Path(root)
|
|
||||||
for path, content in self._files.items():
|
|
||||||
file_path = root / path
|
|
||||||
file_path.parent.mkdir(parents=True, exist_ok=True)
|
file_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
file_path.write_text(content, encoding="utf-8")
|
file_path.write_text(content, encoding="utf-8")
|
||||||
|
|
||||||
def __repr__(self) -> str: # pragma: no cover
|
def __repr__(self) -> str:
|
||||||
return f"VirtualFileSystem({len(self._files)} files)"
|
return f"VirtualFileSystem(files={list(self.files.keys())})"
|
||||||
|
|
||||||
|
|
||||||
# Singleton instance used by the tools.
|
|
||||||
virtual_fs = VirtualFileSystem()
|
|
||||||
|
|||||||
Reference in New Issue
Block a user