mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-28 12:34:50 +02:00
[eric] scheduled tasks: agent can now manage workflows itself + tons of safety, cross-platform polish, and edge-case fixes
This commit is contained in:
ciregenz
@@ -7,11 +7,10 @@ from debugger_backend.DEFAULTS import DEFAULT_COLOR, DEFAULT_TOGGLED, DEFAULT_SE
|
||||
from debugger_backend.path_mngr import get_abspath, get_root_rel_path
|
||||
|
||||
class Directory:
|
||||
def __init__(self, path, color=DEFAULT_COLOR, is_toggled=DEFAULT_TOGGLED,
|
||||
def __init__(self, path, color=DEFAULT_COLOR, is_toggled=DEFAULT_TOGGLED,
|
||||
set_manually=DEFAULT_SET_MANUALLY, emoji=DEFAULT_EMOJI):
|
||||
self.path = path
|
||||
# print(f"Directory init: {self.path}")
|
||||
self.children = [] # Can contain DebugFile or other Directory objects
|
||||
self.children = []
|
||||
self.color = color
|
||||
self.is_toggled = is_toggled
|
||||
self.set_manually = set_manually
|
||||
@@ -24,39 +23,26 @@ class Directory:
|
||||
return get_abspath(self.path)
|
||||
|
||||
def add_child(self, child):
|
||||
"""
|
||||
Adds a child to the directory (either a DebugFile or another Directory).
|
||||
"""
|
||||
"""Append a child DebugFile/Directory."""
|
||||
self.children.append(child)
|
||||
|
||||
def get_ordered_abspaths_and_instances(self):
|
||||
# print("[get_ordered_abspaths]: START")
|
||||
curr_file_path = os.path.abspath(__file__)
|
||||
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(curr_file_path)))
|
||||
# print(f"[get_ordered_abspaths]: Curr path: {curr_file_path}")
|
||||
# print(f"[get_ordered_abspaths]: Dir path: {root_dir}")
|
||||
def construct_ordered_abspaths(dir: Directory, ordered_abspaths: list):
|
||||
dir_path = dir.path
|
||||
full_path = os.path.join(root_dir, dir_path)
|
||||
ordered_abspaths.append({"abspath": full_path, "instance": dir})
|
||||
# print(f"\t[construct_ordered_abspaths]: Full path: {full_path}")
|
||||
for child in dir.children:
|
||||
child_abspath = os.path.join(root_dir, child.path).lower()
|
||||
if os.path.isdir(child_abspath):
|
||||
construct_ordered_abspaths(child, ordered_abspaths)
|
||||
elif os.path.isfile(child_abspath):
|
||||
# print(f"\t[construct_ordered_abspaths]: Child is file: {child_abspath}")
|
||||
ordered_abspaths.append({"abspath": child_abspath, "instance": child})
|
||||
else:
|
||||
print(f"\033[38;5;120mEntry is non existent: {child_abspath}\033[0m")
|
||||
# print(f"\t[construct_ordered_abspaths]: Finished for dir: {full_path}")
|
||||
# print(f"\t[construct_ordered_abspaths]: RETURNING FROM DIR: {full_path}")
|
||||
return ordered_abspaths
|
||||
ordered_abspaths_and_instances = construct_ordered_abspaths(self, [])
|
||||
# print("[get_ordered_abspaths]: Finished getting ordered abspaths and instances")
|
||||
# for abspath_and_instance in ordered_abspaths_and_instances:
|
||||
# abspath = abspath_and_instance["abspath"]
|
||||
# print(f"\t[get_ordered_abspaths]: Abspath: {abspath}")
|
||||
ordered_abspaths = [abspath_and_instance["abspath"] for abspath_and_instance in ordered_abspaths_and_instances]
|
||||
ordered_instances = [abspath_and_instance["instance"] for abspath_and_instance in ordered_abspaths_and_instances]
|
||||
return ordered_abspaths, ordered_instances
|
||||
@@ -65,17 +51,12 @@ class Directory:
|
||||
def build_structure(self):
|
||||
print("[build_structure]: START")
|
||||
root_dir = self.get_abspath()
|
||||
# print(f"[build_structure]: Root dir: {root_dir}")
|
||||
excluded_dirs = [".venv", "debugger", "node_modules", ".git", "__pycache__"]
|
||||
project_structure = []
|
||||
|
||||
def construct_project_structure(dir_path: str, parent_dir: Directory):
|
||||
# print(f"[build_structure]: Scanning dir: {dir_path}")
|
||||
with os.scandir(dir_path) as it:
|
||||
for entry in it:
|
||||
# print(f"[build_structure]: Entry: {entry.path}")
|
||||
if any(excluded_dir in entry.path for excluded_dir in excluded_dirs):
|
||||
# print(f"[build_structure]: Excluding {entry.path}")
|
||||
continue
|
||||
root_rel_path = get_root_rel_path(entry.path)
|
||||
if entry.is_dir():
|
||||
@@ -88,16 +69,12 @@ class Directory:
|
||||
parent_dir.add_child(debug_file)
|
||||
else:
|
||||
continue
|
||||
|
||||
construct_project_structure(root_dir, self)
|
||||
# [print(f"[build_structure]: {file}") for file in project_structure]
|
||||
# print(f"[build_structure]: END")
|
||||
|
||||
construct_project_structure(root_dir, self)
|
||||
return
|
||||
|
||||
def to_dict(self):
|
||||
"""
|
||||
Converts the Directory object to a dictionary format, recursively.
|
||||
"""
|
||||
"""Recursively convert the Directory to a dict."""
|
||||
return {
|
||||
"name": os.path.basename(self.path),
|
||||
"color": self.color,
|
||||
@@ -108,22 +85,14 @@ class Directory:
|
||||
}
|
||||
|
||||
def prune_empty(self):
|
||||
# Recursively prune empty directories
|
||||
# Base case) if the current directory has no children, return
|
||||
# Recursive case) for each of the directories in the current directory, call prune_empty
|
||||
# then remove the directory from the children of the current directory if it has no children
|
||||
for child in self.children[:]:
|
||||
if isinstance(child, Directory):
|
||||
# Recursively prune empty subdirectories
|
||||
child.prune_empty()
|
||||
# If the subdirectory is empty after pruning, remove it
|
||||
if len(child.children) == 0:
|
||||
self.children.remove(child)
|
||||
|
||||
|
||||
def propagate_toggled_state(self):
|
||||
"""
|
||||
Propagates the toggled state down the hierarchy.
|
||||
"""
|
||||
"""Propagate the toggled state down the hierarchy."""
|
||||
for child in self.children:
|
||||
if isinstance(child, DebugFile) and not child.set_manually:
|
||||
child.is_toggled = self.is_toggled
|
||||
@@ -132,9 +101,7 @@ class Directory:
|
||||
child.propagate_toggled_state()
|
||||
|
||||
def propagate_color(self, parent_color=DEFAULT_COLOR):
|
||||
"""
|
||||
Propagates the color from parent to children.
|
||||
"""
|
||||
"""Propagate color from parent to children."""
|
||||
if self.color == DEFAULT_COLOR:
|
||||
self.color = lighten_color(parent_color)
|
||||
for child in self.children:
|
||||
@@ -144,9 +111,7 @@ class Directory:
|
||||
child.propagate_color(self.color)
|
||||
|
||||
def load_from_json(self, json_data):
|
||||
"""
|
||||
Loads a directory structure from a JSON file into this Directory instance.
|
||||
"""
|
||||
"""Load a directory structure from JSON into this Directory."""
|
||||
for item in json_data:
|
||||
if 'children' in item:
|
||||
subdir = Directory(
|
||||
@@ -160,7 +125,6 @@ class Directory:
|
||||
subdir.load_from_json(item['children'])
|
||||
self.add_child(subdir)
|
||||
else:
|
||||
# debug_file = DebugFile.from_dict(item, self)
|
||||
debug_file = DebugFile(
|
||||
filename=item['name'],
|
||||
path=os.path.join(self.path, item['name']),
|
||||
@@ -173,9 +137,7 @@ class Directory:
|
||||
self.add_child(debug_file)
|
||||
|
||||
def reset_colors(self):
|
||||
"""
|
||||
Resets the color of all DebugFile and Directory objects in this directory structure to the default color.
|
||||
"""
|
||||
"""Reset every nested color to the default."""
|
||||
self.color = DEFAULT_COLOR
|
||||
for child in self.children:
|
||||
if isinstance(child, DebugFile):
|
||||
@@ -185,9 +147,7 @@ class Directory:
|
||||
|
||||
|
||||
def lighten_color(color, amount=0.1):
|
||||
"""
|
||||
Lightens the given color by the specified amount.
|
||||
"""
|
||||
"""Lighten the given color by amount."""
|
||||
try:
|
||||
color = color.lstrip('#')
|
||||
r, g, b = int(color[:2], 16), int(color[2:4], 16), int(color[4:6], 16)
|
||||
|
||||
Reference in New Issue
Block a user