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