| 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 "<div style='font-family: monospace; color: #a0aec0; padding: 10px;'>No code changes detected.</div>" | |
| html_lines = ["<div style='font-family: Consolas, Monaco, monospace; font-size: 13px; background-color: #1a202c; color: #e2e8f0; border-radius: 8px; padding: 14px; overflow-x: auto; max-height: 450px; border: 1px solid #2d3748;'>"] | |
| for line in diff_lines: | |
| safe_line = line.replace("&", "&").replace("<", "<").replace(">", ">") | |
| if line.startswith("+") and not line.startswith("+++"): | |
| html_lines.append(f"<div style='background-color: rgba(72, 187, 120, 0.2); color: #68d391; padding: 2px 6px; border-left: 3px solid #38a169;'>{safe_line}</div>") | |
| elif line.startswith("-") and not line.startswith("---"): | |
| html_lines.append(f"<div style='background-color: rgba(245, 101, 101, 0.2); color: #fc8181; padding: 2px 6px; border-left: 3px solid #e53e3e;'>{safe_line}</div>") | |
| elif line.startswith("@"): | |
| html_lines.append(f"<div style='color: #63b3ed; font-weight: bold; padding: 4px 0;'>{safe_line}</div>") | |
| else: | |
| html_lines.append(f"<div style='padding: 2px 6px; color: #cbd5e0;'>{safe_line}</div>") | |
| html_lines.append("</div>") | |
| 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 | |