import os
import json
import difflib
import time
from datetime import datetime
import plotly.graph_objects as go
INCIDENTS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "incidents")
os.makedirs(INCIDENTS_DIR, exist_ok=True)
def generate_code_diff_html(original_code: str, patched_code: str) -> str:
orig_lines = str(original_code).splitlines()
patched_lines = str(patched_code).splitlines()
diff = difflib.unified_diff(orig_lines, patched_lines, fromfile="broken_pipeline.py", tofile="healed_pipeline.py", lineterm="")
diff_lines = list(diff)
if not diff_lines:
return "
No code changes detected.
"
html_lines = [""]
for line in diff_lines:
safe_line = line.replace("&", "&").replace("<", "<").replace(">", ">")
if line.startswith("+") and not line.startswith("+++"):
html_lines.append(f"
{safe_line}
")
elif line.startswith("-") and not line.startswith("---"):
html_lines.append(f"
{safe_line}
")
elif line.startswith("@"):
html_lines.append(f"
{safe_line}
")
else:
html_lines.append(f"
{safe_line}
")
html_lines.append("
")
return "\n".join(html_lines)
def save_incident_report(incident_data: dict) -> str:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
incident_id = incident_data.get("id", f"INC-{timestamp}")
filename = f"{incident_id}.json"
filepath = os.path.join(INCIDENTS_DIR, filename)
with open(filepath, "w", encoding="utf-8") as f:
json.dump(incident_data, f, indent=2)
return filepath
def get_all_incidents() -> list:
incidents = []
if not os.path.exists(INCIDENTS_DIR):
return incidents
for filename in sorted(os.listdir(INCIDENTS_DIR), reverse=True):
if filename.endswith(".json"):
filepath = os.path.join(INCIDENTS_DIR, filename)
try:
with open(filepath, "r", encoding="utf-8") as f:
incidents.append(json.load(f))
except Exception as e:
print(f"Error loading incident {filename}: {e}")
return incidents
def create_telemetry_chart(history_metrics: list):
steps = [m.get("step", i) for i, m in enumerate(history_metrics)]
accuracy = [m.get("accuracy", 0) * 100 for m in history_metrics]
loss = [m.get("loss", 0) for m in history_metrics]
memory = [m.get("memory_mb", 0) for m in history_metrics]
fig = go.Figure()
fig.add_trace(go.Scatter(x=steps, y=accuracy, name="Model Accuracy (%)", line=dict(color="#48bb78", width=3), mode="lines+markers", yaxis="y1"))
fig.add_trace(go.Scatter(x=steps, y=loss, name="Training Loss", line=dict(color="#ed8936", width=2, dash="dash"), mode="lines+markers", yaxis="y2"))
fig.add_trace(go.Scatter(x=steps, y=memory, name="Memory (MB)", line=dict(color="#4299e1", width=2, dash="dot"), mode="lines+markers", yaxis="y3"))
fig.update_layout(
template="plotly_dark",
paper_bgcolor="rgba(0,0,0,0)",
plot_bgcolor="rgba(26,32,44,0.6)",
font=dict(family="Inter, sans-serif", color="#e2e8f0"),
height=320,
margin=dict(l=40, r=40, t=40, b=40),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
xaxis=dict(title="Pipeline Step / Iteration", gridcolor="#2d3748"),
yaxis=dict(title=dict(text="Accuracy (%)", font=dict(color="#48bb78")), tickfont=dict(color="#48bb78"), gridcolor="#2d3748"),
yaxis2=dict(title=dict(text="Loss", font=dict(color="#ed8936")), tickfont=dict(color="#ed8936"), overlaying="y", side="right", showgrid=False),
yaxis3=dict(title=dict(text="Memory (MB)", font=dict(color="#4299e1")), tickfont=dict(color="#4299e1"), overlaying="y", side="right", position=0.95, showgrid=False)
)
return fig