Add export command for JSON and CSV output

src/vmstat_visualizer/cli/__main__.py

@@ -1,9 +1,10 @@
 """
-This is the entrypoint to netcdfella app.
+Entrypoint for the vmstat-visualizer CLI.
 """
 import click
 
 import vmstat_visualizer.cli.visualizer as viz
+from vmstat_visualizer.cli.export import export
 
 
 @click.group(no_args_is_help=True)
@@ -17,6 +18,7 @@ def main():
     "main is the entrypoint of vmstat-visualizer"
     cli.add_command(viz.visualize)
     cli.add_command(viz.compare)
+    cli.add_command(export)
     cli(prog_name="vmstat-visualizer")
 
 

src/vmstat_visualizer/cli/export.py

@@ -0,0 +1,42 @@
+"""CLI command for exporting parsed vmstat data to JSON/CSV."""
+
+import click
+from vmstat_visualizer.parser.parser import Parser
+from vmstat_visualizer.exporters import export_ndjson, export_csv
+
+
+@click.command("export", no_args_is_help=True)
+@click.argument("file", type=click.Path(exists=True))
+@click.option(
+    "--to",
+    "fmt",
+    type=click.Choice(["json", "csv"], case_sensitive=False),
+    required=True,
+    help="Output format: json (NDJSON) or csv.",
+)
+@click.option(
+    "-o",
+    "--output-file",
+    default=None,
+    help="Write output to a file instead of stdout.",
+)
+def export(file, fmt, output_file):
+    """Export parsed vmstat data to JSON or CSV format."""
+    parser = Parser(file)
+    parser.parse()
+
+    if not parser.timeseries:
+        click.echo("No data found in file.", err=True)
+        raise SystemExit(1)
+
+    if fmt == "json":
+        result = export_ndjson(parser.timeseries)
+    else:
+        result = export_csv(parser.timeseries)
+
+    if output_file:
+        with open(output_file, "w") as f:
+            f.write(result)
+        click.echo(f">>> Exported {len(parser.timeseries)} entries to {output_file}")
+    else:
+        click.echo(result)

src/vmstat_visualizer/exporters.py

@@ -0,0 +1,64 @@
+"""Export parsed vmstat data to JSON and CSV formats."""
+
+import csv
+import io
+import json
+
+
+FIELD_ORDER = [
+    ("time", "time"),
+    ("run_queue", "r"),
+    ("blocked_processes", "b"),
+    ("swapped_memory_kb", "swpd"),
+    ("free_memory_kb", "free"),
+    ("inactive_memory_kb", "inact"),
+    ("active_memory_kb", "active"),
+    ("swap_in_kb", "si"),
+    ("swap_out_kb", "so"),
+    ("blocks_in", "bi"),
+    ("blocks_out", "bo"),
+    ("interrupts", "in"),
+    ("context_switches", "cs"),
+    ("user_cpu_percent", "us"),
+    ("system_cpu_percent", "sy"),
+    ("idle_cpu_percent", "id"),
+    ("wait_cpu_percent", "wa"),
+    ("steal_cpu_percent", "st"),
+    ("guest_cpu_percent", "gu"),
+]
+
+
+def _entry_to_dict(ts_entry):
+    """Convert a Timeseries entry to a flat dict, omitting None values."""
+    row = {}
+    for attr, short_name in FIELD_ORDER:
+        val = getattr(ts_entry, attr, None)
+        if val is not None:
+            row[short_name] = val
+    return row
+
+
+def export_ndjson(timeseries):
+    """Export timeseries as newline-delimited JSON (NDJSON)."""
+    lines = []
+    for entry in timeseries:
+        lines.append(json.dumps(_entry_to_dict(entry)))
+    return "\n".join(lines)
+
+
+def export_csv(timeseries):
+    """Export timeseries as CSV."""
+    if not timeseries:
+        return ""
+
+    all_rows = [_entry_to_dict(entry) for entry in timeseries]
+    fieldnames = list(dict.fromkeys(
+        key for row in all_rows for key in row.keys()
+    ))
+
+    output = io.StringIO()
+    writer = csv.DictWriter(output, fieldnames=fieldnames)
+    writer.writeheader()
+    for row in all_rows:
+        writer.writerow(row)
+    return output.getvalue()