mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-10 00:28:31 +00:00
feat: add docstring (#1819)
Co-authored-by: João Moura <joaomdmoura@gmail.com>
This commit is contained in:
committed by
GitHub
parent
45b802a625
commit
a548463fae
@@ -95,7 +95,6 @@ def start(condition: Optional[Union[str, dict, Callable]] = None) -> Callable:
|
|||||||
|
|
||||||
return decorator
|
return decorator
|
||||||
|
|
||||||
|
|
||||||
def listen(condition: Union[str, dict, Callable]) -> Callable:
|
def listen(condition: Union[str, dict, Callable]) -> Callable:
|
||||||
"""
|
"""
|
||||||
Creates a listener that executes when specified conditions are met.
|
Creates a listener that executes when specified conditions are met.
|
||||||
@@ -198,7 +197,6 @@ def router(condition: Union[str, dict, Callable]) -> Callable:
|
|||||||
"""
|
"""
|
||||||
def decorator(func):
|
def decorator(func):
|
||||||
func.__is_router__ = True
|
func.__is_router__ = True
|
||||||
# Handle conditions like listen/start
|
|
||||||
if isinstance(condition, str):
|
if isinstance(condition, str):
|
||||||
func.__trigger_methods__ = [condition]
|
func.__trigger_methods__ = [condition]
|
||||||
func.__condition_type__ = "OR"
|
func.__condition_type__ = "OR"
|
||||||
@@ -220,7 +218,6 @@ def router(condition: Union[str, dict, Callable]) -> Callable:
|
|||||||
|
|
||||||
return decorator
|
return decorator
|
||||||
|
|
||||||
|
|
||||||
def or_(*conditions: Union[str, dict, Callable]) -> dict:
|
def or_(*conditions: Union[str, dict, Callable]) -> dict:
|
||||||
"""
|
"""
|
||||||
Combines multiple conditions with OR logic for flow control.
|
Combines multiple conditions with OR logic for flow control.
|
||||||
|
|||||||
@@ -18,6 +18,8 @@ from crewai.flow.visualization_utils import (
|
|||||||
|
|
||||||
|
|
||||||
class FlowPlot:
|
class FlowPlot:
|
||||||
|
"""Handles the creation and rendering of flow visualization diagrams."""
|
||||||
|
|
||||||
def __init__(self, flow):
|
def __init__(self, flow):
|
||||||
"""
|
"""
|
||||||
Initialize FlowPlot with a flow object.
|
Initialize FlowPlot with a flow object.
|
||||||
|
|||||||
@@ -6,6 +6,8 @@ from crewai.flow.path_utils import safe_path_join, validate_path_exists
|
|||||||
|
|
||||||
|
|
||||||
class HTMLTemplateHandler:
|
class HTMLTemplateHandler:
|
||||||
|
"""Handles HTML template processing and generation for flow visualization diagrams."""
|
||||||
|
|
||||||
def __init__(self, template_path, logo_path):
|
def __init__(self, template_path, logo_path):
|
||||||
"""
|
"""
|
||||||
Initialize HTMLTemplateHandler with validated template and logo paths.
|
Initialize HTMLTemplateHandler with validated template and logo paths.
|
||||||
@@ -29,19 +31,23 @@ class HTMLTemplateHandler:
|
|||||||
raise ValueError(f"Invalid template or logo path: {e}")
|
raise ValueError(f"Invalid template or logo path: {e}")
|
||||||
|
|
||||||
def read_template(self):
|
def read_template(self):
|
||||||
|
"""Read and return the HTML template file contents."""
|
||||||
with open(self.template_path, "r", encoding="utf-8") as f:
|
with open(self.template_path, "r", encoding="utf-8") as f:
|
||||||
return f.read()
|
return f.read()
|
||||||
|
|
||||||
def encode_logo(self):
|
def encode_logo(self):
|
||||||
|
"""Convert the logo SVG file to base64 encoded string."""
|
||||||
with open(self.logo_path, "rb") as logo_file:
|
with open(self.logo_path, "rb") as logo_file:
|
||||||
logo_svg_data = logo_file.read()
|
logo_svg_data = logo_file.read()
|
||||||
return base64.b64encode(logo_svg_data).decode("utf-8")
|
return base64.b64encode(logo_svg_data).decode("utf-8")
|
||||||
|
|
||||||
def extract_body_content(self, html):
|
def extract_body_content(self, html):
|
||||||
|
"""Extract and return content between body tags from HTML string."""
|
||||||
match = re.search("<body.*?>(.*?)</body>", html, re.DOTALL)
|
match = re.search("<body.*?>(.*?)</body>", html, re.DOTALL)
|
||||||
return match.group(1) if match else ""
|
return match.group(1) if match else ""
|
||||||
|
|
||||||
def generate_legend_items_html(self, legend_items):
|
def generate_legend_items_html(self, legend_items):
|
||||||
|
"""Generate HTML markup for the legend items."""
|
||||||
legend_items_html = ""
|
legend_items_html = ""
|
||||||
for item in legend_items:
|
for item in legend_items:
|
||||||
if "border" in item:
|
if "border" in item:
|
||||||
@@ -69,6 +75,7 @@ class HTMLTemplateHandler:
|
|||||||
return legend_items_html
|
return legend_items_html
|
||||||
|
|
||||||
def generate_final_html(self, network_body, legend_items_html, title="Flow Plot"):
|
def generate_final_html(self, network_body, legend_items_html, title="Flow Plot"):
|
||||||
|
"""Combine all components into final HTML document with network visualization."""
|
||||||
html_template = self.read_template()
|
html_template = self.read_template()
|
||||||
logo_svg_base64 = self.encode_logo()
|
logo_svg_base64 = self.encode_logo()
|
||||||
|
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
|
||||||
def get_legend_items(colors):
|
def get_legend_items(colors):
|
||||||
return [
|
return [
|
||||||
{"label": "Start Method", "color": colors["start"]},
|
{"label": "Start Method", "color": colors["start"]},
|
||||||
|
|||||||
@@ -55,6 +55,7 @@ def method_calls_crew(method: Any) -> bool:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
class CrewCallVisitor(ast.NodeVisitor):
|
class CrewCallVisitor(ast.NodeVisitor):
|
||||||
|
"""AST visitor to detect .crew() method calls."""
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self.found = False
|
self.found = False
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user