Coverage for src / graphable / cli / rich_cli.py: 91%
138 statements
« prev ^ index » next coverage.py v7.13.3, created at 2026-02-16 21:32 +0000
1from pathlib import Path
3from rich.console import Console
4from rich.panel import Panel
5from rich.table import Table
6from typer import Argument, Exit, Option, Typer
8from graphable.cli.commands.core import (
9 check_command,
10 checksum_command,
11 convert_command,
12 diff_command,
13 diff_visual_command,
14 info_command,
15 paths_command,
16 reduce_command,
17 render_command,
18 verify_command,
19 write_checksum_command,
20)
21from graphable.cli.commands.serve import serve_command
22from graphable.enums import Engine
24app = Typer(help="Graphable CLI (Rich)")
25console = Console()
28@app.command()
29def info(
30 file: Path = Argument(..., help="Input graph file"),
31 tag: str = Option(None, "--tag", "-t", help="Filter by tag"),
32 upstream_of: str = Option(
33 None, "--upstream-of", help="Filter to ancestors of node"
34 ),
35 downstream_of: str = Option(
36 None, "--downstream-of", help="Filter to descendants of node"
37 ),
38):
39 """Get summary information about a graph."""
40 try:
41 data = info_command(
42 file, tag=tag, upstream_of=upstream_of, downstream_of=downstream_of
43 )
45 title = f"Graph Summary: {file.name}" + (f" (Tag: {tag})" if tag else "")
46 table = Table(title=title)
47 table.add_column("Property", style="cyan")
48 table.add_column("Value", style="magenta")
50 table.add_row("Nodes", str(data["nodes"]))
51 table.add_row("Edges", str(data["edges"]))
52 table.add_row("Sources", ", ".join(data["sources"]))
53 table.add_row("Sinks", ", ".join(data["sinks"]))
55 if data.get("project_duration") is not None:
56 table.add_row("Project Duration", str(data["project_duration"]))
57 table.add_row("Critical Path Length", str(data["critical_path_length"]))
59 console.print(table)
60 except Exception as e:
61 console.print(f"[red]Error:[/red] {e}")
62 raise Exit(1)
65@app.command()
66def check(
67 file: Path = Argument(..., help="Input graph file"),
68 tag: str = Option(None, "--tag", "-t", help="Filter by tag"),
69 upstream_of: str = Option(
70 None, "--upstream-of", help="Filter to ancestors of node"
71 ),
72 downstream_of: str = Option(
73 None, "--downstream-of", help="Filter to descendants of node"
74 ),
75):
76 """Validate graph integrity (cycles and consistency)."""
77 data = check_command(
78 file, tag=tag, upstream_of=upstream_of, downstream_of=downstream_of
79 )
80 if data["valid"]:
81 console.print(
82 Panel("[green]Graph is valid![/green]", title="Validation Result")
83 )
84 else:
85 console.print(
86 Panel(
87 f"[red]Graph is invalid:[/red] {data['error']}",
88 title="Validation Result",
89 )
90 )
91 raise Exit(1)
94@app.command()
95def reduce(
96 input: Path = Argument(..., help="Input graph file"),
97 output: Path = Argument(..., help="Output graph file"),
98 embed: bool = Option(False, "--embed", help="Embed checksum in output"),
99 tag: str = Option(None, "--tag", "-t", help="Filter by tag"),
100 upstream_of: str = Option(
101 None, "--upstream-of", help="Filter to ancestors of node"
102 ),
103 downstream_of: str = Option(
104 None, "--downstream-of", help="Filter to descendants of node"
105 ),
106):
107 """Perform transitive reduction on a graph and save the result."""
108 with console.status("[bold green]Reducing graph..."):
109 reduce_command(
110 input,
111 output,
112 embed_checksum=embed,
113 tag=tag,
114 upstream_of=upstream_of,
115 downstream_of=downstream_of,
116 )
117 console.print(f"[green]Successfully reduced graph and saved to {output}[/green]")
120@app.command()
121def convert(
122 input: Path = Argument(..., help="Input graph file"),
123 output: Path = Argument(..., help="Output graph file"),
124 embed: bool = Option(False, "--embed", help="Embed checksum in output"),
125 tag: str = Option(None, "--tag", "-t", help="Filter by tag"),
126 upstream_of: str = Option(
127 None, "--upstream-of", help="Filter to ancestors of node"
128 ),
129 downstream_of: str = Option(
130 None, "--downstream-of", help="Filter to descendants of node"
131 ),
132):
133 """Convert a graph between supported formats."""
134 with console.status(f"[bold green]Converting {input.name} to {output.name}..."):
135 convert_command(
136 input,
137 output,
138 embed_checksum=embed,
139 tag=tag,
140 upstream_of=upstream_of,
141 downstream_of=downstream_of,
142 )
143 console.print(f"[green]Successfully converted {input} to {output}[/green]")
146@app.command()
147def render(
148 input: Path = Argument(..., help="Input graph file"),
149 output: Path = Argument(..., help="Output image file (.png, .svg)"),
150 engine: Engine = Option(
151 None, "--engine", "-e", help="Rendering engine to use", case_sensitive=False
152 ),
153 tag: str = Option(None, "--tag", "-t", help="Filter by tag"),
154 upstream_of: str = Option(
155 None, "--upstream-of", help="Filter to ancestors of node"
156 ),
157 downstream_of: str = Option(
158 None, "--downstream-of", help="Filter to descendants of node"
159 ),
160):
161 """Render a graph as an image."""
162 with console.status(f"[bold green]Rendering {input.name} to {output.name}..."):
163 try:
164 render_command(
165 input,
166 output,
167 engine=engine,
168 tag=tag,
169 upstream_of=upstream_of,
170 downstream_of=downstream_of,
171 )
172 console.print(f"[green]Successfully rendered {input} to {output}[/green]")
173 except Exception as e:
174 console.print(f"[red]Error:[/red] {e}")
175 raise Exit(1)
178@app.command()
179def checksum(
180 file: Path = Argument(..., help="Graph file"),
181 tag: str = Option(None, "--tag", "-t", help="Filter by tag"),
182 upstream_of: str = Option(
183 None, "--upstream-of", help="Filter to ancestors of node"
184 ),
185 downstream_of: str = Option(
186 None, "--downstream-of", help="Filter to descendants of node"
187 ),
188):
189 """Calculate and print the graph checksum."""
190 try:
191 console.print(
192 checksum_command(
193 file, tag=tag, upstream_of=upstream_of, downstream_of=downstream_of
194 )
195 )
196 except Exception as e:
197 console.print(f"[red]Error:[/red] {e}")
198 raise Exit(1)
201@app.command()
202def verify(
203 file: Path = Argument(..., help="Graph file"),
204 expected: str = Option(None, "--expected", help="Expected checksum (hex)"),
205 tag: str = Option(None, "--tag", "-t", help="Filter by tag"),
206 upstream_of: str = Option(
207 None, "--upstream-of", help="Filter to ancestors of node"
208 ),
209 downstream_of: str = Option(
210 None, "--downstream-of", help="Filter to descendants of node"
211 ),
212):
213 """Verify graph checksum (embedded or provided)."""
214 try:
215 data = verify_command(
216 file,
217 expected,
218 tag=tag,
219 upstream_of=upstream_of,
220 downstream_of=downstream_of,
221 )
222 if data["valid"] is True:
223 console.print("[green]Checksum verified successfully.[/green]")
224 elif data["valid"] is False:
225 console.print(
226 f"[red]Checksum mismatch![/red] Expected {data['expected']}, got {data['actual']}"
227 )
228 raise Exit(1)
229 else:
230 console.print(
231 f"[yellow]No checksum found to verify.[/yellow] Current hash: {data['actual']}"
232 )
233 except Exception as e:
234 console.print(f"[red]Error:[/red] {e}")
235 raise Exit(1)
238@app.command()
239def write_checksum(
240 file: Path = Argument(..., help="Graph file"),
241 output: Path = Argument(..., help="Output checksum file"),
242 tag: str = Option(None, "--tag", "-t", help="Filter by tag"),
243 upstream_of: str = Option(
244 None, "--upstream-of", help="Filter to ancestors of node"
245 ),
246 downstream_of: str = Option(
247 None, "--downstream-of", help="Filter to descendants of node"
248 ),
249):
250 """Write graph checksum to a standalone file."""
251 try:
252 write_checksum_command(
253 file,
254 output,
255 tag=tag,
256 upstream_of=upstream_of,
257 downstream_of=downstream_of,
258 )
259 console.print(f"[green]Checksum written to {output}[/green]")
260 except Exception as e:
261 console.print(f"[red]Error:[/red] {e}")
262 raise Exit(1)
265@app.command()
266def diff(
267 file1: Path = Argument(..., help="First graph file"),
268 file2: Path = Argument(..., help="Second graph file"),
269 output: Path = Option(None, "--output", "-o", help="Output file for visual diff"),
270 tag: str = Option(None, "--tag", "-t", help="Filter by tag"),
271):
272 """Compare two graphs and highlight differences."""
273 try:
274 if output:
275 diff_visual_command(file1, file2, output, tag=tag)
276 console.print(f"[green]Visual diff saved to {output}[/green]")
277 return
279 data = diff_command(file1, file2, tag=tag)
280 # ... (rest of diff command)
282 if not any(data.values()):
283 console.print("[green]Graphs are identical.[/green]")
284 return
286 table = Table(title=f"Graph Diff: {file1.name} vs {file2.name}")
287 table.add_column("Category", style="cyan")
288 table.add_column("Changes", style="magenta")
290 if data["added_nodes"]:
291 table.add_row("Added Nodes", ", ".join(map(str, data["added_nodes"])))
292 if data["removed_nodes"]:
293 table.add_row("Removed Nodes", ", ".join(map(str, data["removed_nodes"])))
294 if data["modified_nodes"]:
295 table.add_row("Modified Nodes", ", ".join(map(str, data["modified_nodes"])))
297 if data["added_edges"]:
298 table.add_row(
299 "Added Edges",
300 ", ".join(f"{u}->{v}" for u, v in data["added_edges"]),
301 )
302 if data["removed_edges"]:
303 table.add_row(
304 "Removed Edges",
305 ", ".join(f"{u}->{v}" for u, v in data["removed_edges"]),
306 )
307 if data["modified_edges"]:
308 table.add_row(
309 "Modified Edges",
310 ", ".join(f"{u}->{v}" for u, v in data["modified_edges"]),
311 )
313 console.print(table)
314 except Exception as e:
315 console.print(f"[red]Error:[/red] {e}")
316 raise Exit(1)
319@app.command()
320def paths(
321 file: Path = Argument(..., help="Graph file"),
322 source: str = Argument(..., help="Source node reference"),
323 target: str = Argument(..., help="Target node reference"),
324 tag: str = Option(None, "--tag", "-t", help="Filter by tag"),
325 upstream_of: str = Option(
326 None, "--upstream-of", help="Filter to ancestors of node"
327 ),
328 downstream_of: str = Option(
329 None, "--downstream-of", help="Filter to descendants of node"
330 ),
331):
332 """Find all paths between two nodes."""
333 try:
334 data = paths_command(
335 file,
336 source,
337 target,
338 tag=tag,
339 upstream_of=upstream_of,
340 downstream_of=downstream_of,
341 )
343 if not data:
344 console.print(
345 f"[yellow]No paths found from '{source}' to '{target}'.[/yellow]"
346 )
347 return
349 title = f"Paths from '{source}' to '{target}'"
350 console.print(Panel(f"[bold]{title}[/bold]"))
352 for i, path in enumerate(data, 1):
353 console.print(f"{i}. {' -> '.join(path)}")
354 except Exception as e:
355 console.print(f"[red]Error:[/red] {e}")
356 raise Exit(1)
359@app.command()
360def serve(
361 file: Path = Argument(..., help="Graph file to serve"),
362 port: int = Option(8000, "--port", "-p", help="Port to serve on"),
363):
364 """Start a local web server with live-reloading visualization."""
365 try:
366 console.print(f"[green]Serving {file} on http://127.0.0.1:{port}[/green]")
367 console.print("[yellow]Press Ctrl+C to stop.[/yellow]")
368 serve_command(file, port=port)
369 except Exception as e:
370 console.print(f"[red]Error:[/red] {e}")
371 raise Exit(1)
374if __name__ == "__main__":
375 app()