From 0db5ce931840a5095e1b40cd139b182556d0576e Mon Sep 17 00:00:00 2001 From: omercengiz Date: Wed, 2 Sep 2026 00:30:36 +0300 Subject: [PATCH] feat: add richer CLI reporting --- README.md | 6 +- src/flowsense/cli/main.py | 91 +-------------- src/flowsense/cli/report.py | 218 ++++++++++++++++++++++++++++++++++++ tests/test_cli_report.py | 135 ++++++++++++++++++++++ 4 files changed, 360 insertions(+), 90 deletions(-) create mode 100644 src/flowsense/cli/report.py create mode 100644 tests/test_cli_report.py diff --git a/README.md b/README.md index 83dca55..c14cd54 100644 --- a/README.md +++ b/README.md @@ -135,6 +135,11 @@ Then run: flowsense analyze ``` +The CLI report includes a DAG summary and separate tables for task drift, +handoff drift, change points, trends, propagation paths, and diagnostics. +Results are ordered by severity or subject so repeated analyses remain easy to +compare. + ## Library API FlowSense can also be used as a Python library through its supported top-level @@ -283,7 +288,6 @@ The current implementation should be considered experimental and is not yet inte Planned areas include: - DAG-level analysis models -- richer CLI reporting - broader Airflow compatibility testing ## License diff --git a/src/flowsense/cli/main.py b/src/flowsense/cli/main.py index fe15a91..071ff9e 100644 --- a/src/flowsense/cli/main.py +++ b/src/flowsense/cli/main.py @@ -4,9 +4,9 @@ import typer from rich.console import Console -from rich.table import Table from flowsense.application import analyze_dag +from flowsense.cli.report import render_analysis from flowsense.domain import AnalysisPolicy, MappedTaskAggregation from flowsense.infrastructure.airflow import AirflowApiError, AirflowClient @@ -63,91 +63,4 @@ def analyze( console.print(f"[bold red]Airflow request failed:[/bold red] {exc}") raise typer.Exit(code=1) from exc - console.print(f"\n[bold]FlowSense Analysis — {analysis.dag_id}[/bold]\n") - - table = Table() - - table.add_column("Task") - table.add_column("Baseline") - table.add_column("Current") - table.add_column("Deviation") - table.add_column("Z-Score") - table.add_column("Severity") - table.add_column("Impact") - - for task_id, result in analysis.drift_results.items(): - impact = analysis.task_impacts.get(task_id) - impact_label = impact.classification if impact else "-" - - table.add_row( - task_id, - f"{result.baseline:.2f}s", - f"{result.current:.2f}s", - f"{result.deviation_percent:+.1f}%", - f"{result.robust_z_score:.2f}", - result.severity, - impact_label, - ) - - console.print(table) - - console.print(f"\nOverall Severity: [bold]{analysis.overall_severity}[/bold]") - - if analysis.change_point_results or analysis.handoff_change_point_results: - console.print("\n[bold]Change Points[/bold]\n") - - for result in [ - *analysis.change_point_results.values(), - *analysis.handoff_change_point_results.values(), - ]: - change = ( - f"{result.change_percent:+.1f}%" - if result.change_percent is not None - else "n/a" - ) - console.print( - f"{result.subject_id}: {result.direction} at observation " - f"{result.change_index + 1} ({change}, score={result.score:.2f})" - ) - - if analysis.trend_results or analysis.handoff_trend_results: - console.print("\n[bold]Trends[/bold]\n") - - for result in [ - *analysis.trend_results.values(), - *analysis.handoff_trend_results.values(), - ]: - change = ( - f"{result.change_percent:+.1f}%" - if result.change_percent is not None - else "n/a" - ) - console.print( - f"{result.subject_id}: {result.direction} " - f"({result.slope_per_observation:+.2f}/run, {change}, " - f"score={result.score:.2f})" - ) - - if analysis.primary_origin: - console.print(f"Primary Origin: [bold]{analysis.primary_origin.task_id}[/bold]") - console.print(f"Reason: [bold]{analysis.primary_origin.classification}[/bold]") - console.print(f"Severity: [bold]{analysis.primary_origin.severity}[/bold]") - console.print( - f"Propagation Score: {analysis.primary_origin.propagation_score:.2f}" - ) - - if analysis.propagation_results: - console.print("\n[bold]Propagation Analysis[/bold]\n") - - for result in analysis.propagation_results: - console.print(f"Origin: {result.origin_task}") - console.print(f"Path: {' -> '.join(result.path)}") - console.print(f"Propagation Score: {result.propagation_score:.2f}") - - if analysis.diagnostics: - console.print("\n[bold yellow]Diagnostics[/bold yellow]\n") - - for diagnostic in analysis.diagnostics: - console.print( - f"[{diagnostic.code}] {diagnostic.subject_id}: {diagnostic.message}" - ) + render_analysis(console, analysis) diff --git a/src/flowsense/cli/report.py b/src/flowsense/cli/report.py new file mode 100644 index 0000000..bf20750 --- /dev/null +++ b/src/flowsense/cli/report.py @@ -0,0 +1,218 @@ +from __future__ import annotations + +from rich.console import Console +from rich.panel import Panel +from rich.table import Table +from rich.text import Text + +from flowsense.domain import DAGAnalysis, Severity +from flowsense.domain.enums import SEVERITY_SCORE + +_SEVERITY_STYLES = { + Severity.NORMAL: "green", + Severity.MEDIUM: "yellow", + Severity.HIGH: "bright_red", + Severity.CRITICAL: "bold red", +} + + +def _severity_text(severity: Severity) -> Text: + return Text(str(severity), style=_SEVERITY_STYLES[severity]) + + +def _percent(value: float | None) -> str: + return f"{value:+.1f}%" if value is not None else "n/a" + + +def _render_summary(console: Console, analysis: DAGAnalysis) -> None: + summary = Table.grid(padding=(0, 2)) + summary.add_column(style="bold") + summary.add_column() + summary.add_row("DAG", analysis.dag_id) + summary.add_row("Runs analyzed", str(analysis.runs_analyzed)) + summary.add_row("Overall severity", _severity_text(analysis.overall_severity)) + + if analysis.primary_origin is not None: + summary.add_row("Primary origin", analysis.primary_origin.task_id) + summary.add_row("Classification", str(analysis.primary_origin.classification)) + summary.add_row( + "Propagation score", + f"{analysis.primary_origin.propagation_score:.2f}", + ) + + console.print(Panel(summary, title="FlowSense Analysis", expand=False)) + + +def _render_task_drift(console: Console, analysis: DAGAnalysis) -> None: + table = Table(title="Task Drift") + table.add_column("Task") + table.add_column("Baseline", justify="right") + table.add_column("Current", justify="right") + table.add_column("Deviation", justify="right") + table.add_column("Z-Score", justify="right") + table.add_column("Severity") + table.add_column("Impact") + + ordered_results = sorted( + analysis.drift_results.items(), + key=lambda item: (-SEVERITY_SCORE[item[1].severity], item[0]), + ) + + for task_id, result in ordered_results: + impact = analysis.task_impacts.get(task_id) + table.add_row( + task_id, + f"{result.baseline:.2f}s", + f"{result.current:.2f}s", + f"{result.deviation_percent:+.1f}%", + f"{result.robust_z_score:.2f}", + _severity_text(result.severity), + str(impact.classification) if impact else "-", + ) + + console.print(table) + + +def _render_handoff_drift(console: Console, analysis: DAGAnalysis) -> None: + if not analysis.handoff_drift_results: + return + + table = Table(title="Handoff Drift") + table.add_column("Edge") + table.add_column("Baseline", justify="right") + table.add_column("Current", justify="right") + table.add_column("Deviation", justify="right") + table.add_column("Z-Score", justify="right") + table.add_column("Severity") + + ordered_results = sorted( + analysis.handoff_drift_results.items(), + key=lambda item: (-SEVERITY_SCORE[item[1].severity], item[0]), + ) + + for (upstream, downstream), result in ordered_results: + table.add_row( + f"{upstream} -> {downstream}", + f"{result.baseline:.2f}s", + f"{result.current:.2f}s", + f"{result.deviation_percent:+.1f}%", + f"{result.robust_z_score:.2f}", + _severity_text(result.severity), + ) + + console.print(table) + + +def _render_change_points(console: Console, analysis: DAGAnalysis) -> None: + results = [ + *analysis.change_point_results.values(), + *analysis.handoff_change_point_results.values(), + ] + if not results: + return + + table = Table(title="Change Points") + table.add_column("Subject") + table.add_column("Direction") + table.add_column("Observation", justify="right") + table.add_column("Before", justify="right") + table.add_column("After", justify="right") + table.add_column("Change", justify="right") + table.add_column("Score", justify="right") + + for result in sorted(results, key=lambda item: item.subject_id): + table.add_row( + result.subject_id, + str(result.direction), + str(result.change_index + 1), + f"{result.before_median:.2f}s", + f"{result.after_median:.2f}s", + _percent(result.change_percent), + f"{result.score:.2f}", + ) + + console.print(table) + + +def _render_trends(console: Console, analysis: DAGAnalysis) -> None: + results = [ + *analysis.trend_results.values(), + *analysis.handoff_trend_results.values(), + ] + if not results: + return + + table = Table(title="Trends") + table.add_column("Subject") + table.add_column("Direction") + table.add_column("Slope / run", justify="right") + table.add_column("Est. change", justify="right") + table.add_column("Change", justify="right") + table.add_column("Consistency", justify="right") + table.add_column("Score", justify="right") + + for result in sorted(results, key=lambda item: item.subject_id): + table.add_row( + result.subject_id, + str(result.direction), + f"{result.slope_per_observation:+.2f}s", + f"{result.estimated_change:+.2f}s", + _percent(result.change_percent), + f"{result.directional_consistency:.0%}", + f"{result.score:.2f}", + ) + + console.print(table) + + +def _render_propagation(console: Console, analysis: DAGAnalysis) -> None: + if not analysis.propagation_results: + return + + table = Table(title="Propagation") + table.add_column("Origin") + table.add_column("Path") + table.add_column("Affected", justify="right") + table.add_column("Score", justify="right") + + for result in sorted( + analysis.propagation_results, + key=lambda item: (item.origin_task, item.path), + ): + table.add_row( + result.origin_task, + " -> ".join(result.path), + str(len(result.affected_tasks)), + f"{result.propagation_score:.2f}", + ) + + console.print(table) + + +def _render_diagnostics(console: Console, analysis: DAGAnalysis) -> None: + if not analysis.diagnostics: + return + + table = Table(title="Diagnostics", title_style="bold yellow") + table.add_column("Code", style="yellow") + table.add_column("Subject") + table.add_column("Message") + + for diagnostic in sorted( + analysis.diagnostics, + key=lambda item: (item.code, item.subject_id), + ): + table.add_row(diagnostic.code, diagnostic.subject_id, diagnostic.message) + + console.print(table) + + +def render_analysis(console: Console, analysis: DAGAnalysis) -> None: + """Render a complete human-readable analysis report.""" + _render_summary(console, analysis) + _render_task_drift(console, analysis) + _render_handoff_drift(console, analysis) + _render_change_points(console, analysis) + _render_trends(console, analysis) + _render_propagation(console, analysis) + _render_diagnostics(console, analysis) diff --git a/tests/test_cli_report.py b/tests/test_cli_report.py new file mode 100644 index 0000000..8182754 --- /dev/null +++ b/tests/test_cli_report.py @@ -0,0 +1,135 @@ +from io import StringIO + +from rich.console import Console + +from flowsense.cli.report import render_analysis +from flowsense.domain import ( + AnalysisDiagnostic, + ChangePointResult, + DAGAnalysis, + DriftResult, + PropagationResult, + RootCauseResult, + TaskImpact, + TrendResult, +) + + +def test_renders_complete_analysis_report() -> None: + task_drift = DriftResult( + task_id="transform", + baseline=3.0, + current=9.0, + mad=0.2, + robust_z_score=8.0, + deviation_percent=200.0, + severity="CRITICAL", + ) + handoff_drift = DriftResult( + task_id="extract->transform", + baseline=1.0, + current=4.0, + mad=0.1, + robust_z_score=7.0, + deviation_percent=300.0, + severity="HIGH", + ) + analysis = DAGAnalysis( + dag_id="demo", + runs_analyzed=8, + overall_severity="CRITICAL", + primary_origin=RootCauseResult( + task_id="transform", + classification="OWN_DRIFT", + severity="CRITICAL", + propagation_score=0.75, + ), + drift_results={"transform": task_drift}, + handoff_drift_results={("extract", "transform"): handoff_drift}, + task_impacts={ + "transform": TaskImpact( + task_id="transform", + classification="COMBINED", + task_severity="CRITICAL", + upstream_handoff_severity="HIGH", + ) + }, + propagation_results=[ + PropagationResult( + origin_task="transform", + affected_tasks=["load"], + path=["transform", "load"], + propagation_score=0.75, + ) + ], + dependencies={"transform": ["load"], "load": []}, + diagnostics=[ + AnalysisDiagnostic( + code="INSUFFICIENT_TASK_HISTORY", + subject_id="load", + message="Not enough observations.", + ) + ], + change_point_results={ + "transform": ChangePointResult( + subject_id="transform", + change_index=4, + before_median=3.0, + after_median=6.0, + change_percent=100.0, + score=5.0, + direction="INCREASE", + ) + }, + trend_results={ + "transform": TrendResult( + subject_id="transform", + direction="INCREASING", + slope_per_observation=1.0, + estimated_change=7.0, + change_percent=233.3, + score=5.0, + directional_consistency=1.0, + observations=8, + ) + }, + ) + output = StringIO() + console = Console(file=output, width=160, color_system=None) + + render_analysis(console, analysis) + + report = output.getvalue() + assert "FlowSense Analysis" in report + assert "Task Drift" in report + assert "Handoff Drift" in report + assert "Change Points" in report + assert "Trends" in report + assert "Propagation" in report + assert "Diagnostics" in report + assert "transform -> load" in report + assert "INSUFFICIENT_TASK_HISTORY" in report + + +def test_renders_empty_task_drift_table_without_optional_sections() -> None: + analysis = DAGAnalysis( + dag_id="empty", + runs_analyzed=0, + overall_severity="NORMAL", + primary_origin=None, + drift_results={}, + handoff_drift_results={}, + task_impacts={}, + propagation_results=[], + dependencies={}, + ) + output = StringIO() + console = Console(file=output, width=120, color_system=None) + + render_analysis(console, analysis) + + report = output.getvalue() + assert "FlowSense Analysis" in report + assert "Task Drift" in report + assert "Handoff Drift" not in report + assert "Diagnostics" not in report