This tool is a small command-line program for plotting live numeric data directly inside a terminal. It reads values from standard input, sends them to gnuplot, and displays a simple text-based graph that updates as new data arrives.
Purpose
The purpose of this tool is to make stream data easy to observe without opening a separate graphical window.
It is useful when you want to monitor changing values in real time, such as sensor readings, log output, benchmark results, or values produced by another command-line program.
Because the graph is drawn inside the terminal, it can be used in simple shell workflows and remote environments where a graphical desktop may not be available.
How It Works
The program reads lines from standard input. Each line is expected to contain numeric data.
For every valid numeric value, the program records the time since startup and plots the value against elapsed time. The graph is updated automatically whenever new data is received.
Only a limited number of recent points are kept. By default, the tool stores up to 200 points, which keeps the display focused on recent changes.
Input Format
The tool accepts comma-separated or tab-separated input.
For example, this input can be plotted:
1.2
1.5
1.8
2.1
It can also read the first column from CSV-like data:
1.2,ok
1.5,ok
1.8,ok
2.1,ok
If a line does not contain a valid numeric value, it is ignored.
Basic Usage
Save the script as a file, for example:
stream-plot.py
Make it executable:
chmod +x stream-plot.py
Then pipe numeric data into it:
python3 generate_values.py | ./stream-plot.py
A simple live plot will appear in the terminal.
Example
A simple test can be done with a shell loop:
while true; do
echo $RANDOM
sleep 1
done | ./stream-plot.py
This sends a new random value every second. The terminal plot updates each time a new value is received.
Terminal Resizing
The tool detects terminal size and adjusts the plot area. If the terminal window is resized, the plot size is updated so the graph continues to fit the available space.
Stopping the Tool
The tool can be stopped with:
Ctrl+C
It handles shutdown by closing the gnuplot process cleanly.
Requirements
This tool requires:
gnuplot
It also uses Python 3 and standard Python libraries such as asyncio, csv, and signal.
Summary
This terminal stream plotting tool provides a simple way to watch live numeric data from the command line. It is lightweight, works with standard input, and is suitable for quick monitoring tasks where a full plotting application is unnecessary.
#!/usr/bin/env python3
import asyncio
import csv
import io
import shutil
import signal
import sys
import time
import uuid
from collections import deque
def get_plot_size():
size = shutil.get_terminal_size(fallback=(120, 30))
width = max(20, size.columns)
height = max(10, size.lines - 1)
return width, height
class StreamGnuplot:
def __init__(
self,
*,
max_points=200,
width=120,
height=30,
value_column=0,
# title="stdin stream",
# ylabel="value",
):
self.max_points = max_points
self.width = width
self.height = height
self.value_column = value_column
# self.title = title
# self.ylabel = ylabel
self.start_time = time.monotonic()
self.points = deque(maxlen=max_points)
self.proc = None
self._lock = asyncio.Lock()
self._first_draw = True
self._resize_requested = False
async def start(self):
self.proc = await asyncio.create_subprocess_exec(
"gnuplot",
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
await self._write(self._gnuplot_setup_script())
def request_resize(self):
self._resize_requested = True
def _gnuplot_setup_script(self):
# set title "{self.title}"
# set ylabel "{self.ylabel}"
return f"""
set terminal dumb {self.width} {self.height}
set print "-"
set xlabel "time since start [s]"
set grid
set key off
"""
async def _apply_resize_if_needed(self):
if not self._resize_requested:
return
self._resize_requested = False
new_width, new_height = get_plot_size()
if new_width == self.width and new_height == self.height:
return
self.width = new_width
self.height = new_height
await self._write(f"set terminal dumb {self.width} {self.height}\n")
# ここでは _first_draw を戻さない。
# 「1回目だけ \033[H しない」という条件を維持する。
async def stop(self):
if self.proc is None:
return
try:
await self._write("exit\n")
except Exception:
pass
try:
await asyncio.wait_for(self.proc.wait(), timeout=1.0)
except asyncio.TimeoutError:
self.proc.terminate()
await self.proc.wait()
async def _write(self, text):
if self.proc is None or self.proc.stdin is None:
raise RuntimeError("gnuplot is not running")
self.proc.stdin.write(text.encode("utf-8"))
await self.proc.stdin.drain()
async def add_value(self, value):
elapsed = time.monotonic() - self.start_time
self.points.append((elapsed, value))
await self.replot()
async def replot(self):
if not self.points:
return
async with self._lock:
await self._apply_resize_if_needed()
marker = f"__GPLOT_DONE_{uuid.uuid4().hex}__"
data = "\n".join(
f"{x:.6f} {y:.12g}"
for x, y in self.points
)
script = f"""
plot "-" using 1:2 with lines
{data}
e
print "{marker}"
"""
await self._write(script)
plot_text = await self._read_until_marker(marker)
self._draw_to_terminal(plot_text)
async def _read_until_marker(self, marker):
if self.proc is None or self.proc.stdout is None:
raise RuntimeError("gnuplot stdout is not available")
lines = []
while True:
raw = await self.proc.stdout.readline()
if raw == b"":
raise RuntimeError("gnuplot stdout closed unexpectedly")
line = raw.decode("utf-8", errors="replace")
if line.rstrip("\r\n") == marker:
break
lines.append(line)
return "".join(lines)
def _draw_to_terminal(self, plot_text):
"""
初回は cursor home せず、その場にプロットを出力する。
2回目以降は左上へ戻ってプロットテキストを上書きする。
プロットテキスト最後の改行だけを出力しない。
sys.stdout.write("\\033[J") は使わない。
"""
if self._first_draw:
self._first_draw = False
else:
sys.stdout.write("\033[H")
if plot_text.endswith("\r\n"):
plot_text = plot_text[:-2]
elif plot_text.endswith("\n") or plot_text.endswith("\r"):
plot_text = plot_text[:-1]
sys.stdout.write(plot_text)
sys.stdout.flush()
def detect_delimiter(line):
if "\t" in line:
return "\t"
return ","
def parse_numeric_value(line, value_column):
line = line.strip()
if not line:
return None
delimiter = detect_delimiter(line)
reader = csv.reader(io.StringIO(line), delimiter=delimiter)
row = next(reader, None)
if not row:
return None
try:
return float(row[value_column])
except (ValueError, IndexError):
return None
async def read_stdin_stream(plotter):
loop = asyncio.get_running_loop()
while True:
line = await loop.run_in_executor(None, sys.stdin.readline)
if line == "":
break
value = parse_numeric_value(line, plotter.value_column)
if value is None:
continue
await plotter.add_value(value)
async def drain_stderr(proc):
if proc.stderr is None:
return
while True:
raw = await proc.stderr.readline()
if raw == b"":
break
# sys.stderr.write(raw.decode("utf-8", errors="replace"))
# sys.stderr.flush()
async def main():
width, height = get_plot_size()
plotter = StreamGnuplot(
max_points=200,
width=width,
height=height,
value_column=0,
# title="stdin stream",
# ylabel="value",
)
stop_event = asyncio.Event()
def request_stop():
stop_event.set()
loop = asyncio.get_running_loop()
for sig in (signal.SIGINT, signal.SIGTERM):
try:
loop.add_signal_handler(sig, request_stop)
except NotImplementedError:
pass
try:
loop.add_signal_handler(signal.SIGWINCH, plotter.request_resize)
except (AttributeError, NotImplementedError):
pass
await plotter.start()
stderr_task = asyncio.create_task(drain_stderr(plotter.proc))
reader_task = asyncio.create_task(read_stdin_stream(plotter))
stop_task = asyncio.create_task(stop_event.wait())
done, pending = await asyncio.wait(
{reader_task, stop_task},
return_when=asyncio.FIRST_COMPLETED,
)
for task in pending:
task.cancel()
stderr_task.cancel()
await plotter.stop()
if __name__ == "__main__":
asyncio.run(main())
Top comments (0)