feat(skills): add chart-to-chat skill for dbt-charts and quickchart
This commit is contained in:
Moritz Graf
@@ -78,6 +78,12 @@ Universal QR code generator featuring the Haumdaucher logo in the center.
|
||||
- **Features:** High-redundancy QR code synthesis (`ERROR_CORRECT_H`), logo overlay with clean white badge, and native Telegram image delivery via `MEDIA:<path>`.
|
||||
- **Tools:** Standalone CLI at `scripts/generate_qr.py`.
|
||||
|
||||
### [chart-to-chat](skills/chart-to-chat/SKILL.md)
|
||||
Instant chart and diagram generation for chat and Telegram.
|
||||
- **Features:** Transforms structured data (JSON, CSV, key-value metrics) into publication-grade PNG diagrams (bar, line, area, donut, KPI cards, tables) powered by dbt-charts (`dct`) with QuickChart.io fallback.
|
||||
- **Delivery:** Generates local PNGs and outputs `MEDIA:<path>` for native Telegram image rendering.
|
||||
- **Tools:** Standalone CLI at `scripts/chart.py`.
|
||||
|
||||
### [grill-me](skills/grill-me/SKILL.md)
|
||||
Relentless, structured interview to stress-test and sharpen plans, architectures, decisions, or ideas before writing code.
|
||||
- **Features:** Decision tree & frontier exploration, Telegram/mobile-optimized batching (1-3 questions per round), facts-first environment research, anti-sycophancy challenge, and final specification/ADR synthesis.
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
---
|
||||
name: chart-to-chat
|
||||
description: Generiert hochwertige Diagramme und Charts (Balken, Linien, Verlauf, Donut/Kuchen, KPI-Karten, Tabellen) aus strukturierten Chat-Daten und sendet sie direkt als Bild (PNG) nach Telegram. Basiert auf dbt-charts (lokales Rendering via dct) mit QuickChart.io-Fallback.
|
||||
---
|
||||
|
||||
# 📊 Chart-to-Chat: Diagramm- und Chart-Generator
|
||||
|
||||
Dieser Skill transformiert tabellarische Daten, Verläufe, Statistiken oder Kennzahlen direkt in fertige, publikationsreife **PNG-Grafiken**, die nativ per Telegram-Bildnachricht ausgeliefert werden.
|
||||
|
||||
---
|
||||
|
||||
## 🎯 Wann diesen Skill verwenden?
|
||||
|
||||
* Immer wenn der Nutzer nach einer Visualisierung fragt:
|
||||
* *„Zeig mir das als Diagramm / Chart / Balkendiagramm“*
|
||||
* *„Plotte mir den Verlauf von X über die letzten 7 Tage“*
|
||||
* *„Visualisiere meine Ausgaben / mein Training / meine Schritte“*
|
||||
* *„Erstelle eine Grafik / Kurve dazu“*
|
||||
* Bei Vergleichsdaten oder Verteilungen (z. B. Anteile, Kategorien, Monatsumsätze).
|
||||
* Für KPI-Zusammenfassungen mit Hauptwert und Delta.
|
||||
|
||||
---
|
||||
|
||||
## 📈 Welcher Chart-Typ passt wann?
|
||||
|
||||
| Chart-Typ (`--type`) | Bester Anwendungsfall | Beispiel |
|
||||
|---|---|---|
|
||||
| `bar` (Default) | Diskrete Kategorien, Rankings, Monatsvergleiche | Schritte pro Tag, Stunden pro Projekt |
|
||||
| `hbar` | Viele Kategorien mit langen Namen | Ranking der Supermärkte nach Ersparnis |
|
||||
| `line` | Zeitreihen, kontinuierliche Verläufe, Messwerte | Herzfrequenz, Temperaturverlauf, Gewicht, Watt |
|
||||
| `area` | Kumulierte Volumina oder Trends über Zeit | Datenverbrauch, kumulierter Regen, Trainingsvolumen |
|
||||
| `donut` / `pie` | Prozentuale Anteile an einem Ganzen (100 %) | Ausgaben nach Kategorie, Makronährstoff-Verteilung |
|
||||
| `kpi` | Einzelne prominente Kennzahl mit Kontext/Delta | Aktueller FTP, Monatsumsatz, Gesamtschritte |
|
||||
| `table` | Strukturierte Datentabelle mit formatierten Spalten | Vergleichstabelle von Produkten oder Tarifen |
|
||||
|
||||
---
|
||||
|
||||
## 🛠️ Ausführung
|
||||
|
||||
Führe das Skript `chart.py` via `uv run` aus (Cluster-Pfad: `/home/node/mop-skills/mop-skills/skills/chart-to-chat/scripts/chart.py`):
|
||||
|
||||
### 1. Daten direkt als JSON-String übergeben
|
||||
|
||||
```bash
|
||||
uv run /home/node/mop-skills/mop-skills/skills/chart-to-chat/scripts/chart.py \
|
||||
--title "Wöchentliche Laufdistanz" \
|
||||
--type bar \
|
||||
--data '[{"tag": "Mo", "km": 8.5}, {"tag": "Mi", "km": 12.0}, {"tag": "Fr", "km": 6.2}, {"tag": "So", "km": 16.5}]'
|
||||
```
|
||||
|
||||
### 2. Daten per Pipe (Standard-Eingabe) übergeben
|
||||
|
||||
Besonders empfohlen bei größeren Datenmengen oder um Shell-Escaping-Probleme zu vermeiden:
|
||||
|
||||
```bash
|
||||
cat << 'EOF' | uv run /home/node/mop-skills/mop-skills/skills/chart-to-chat/scripts/chart.py --title "Ruhepuls Trend" --type line --data -
|
||||
[
|
||||
{"datum": "12.09.", "puls": 54},
|
||||
{"datum": "13.09.", "puls": 52},
|
||||
{"datum": "14.09.", "puls": 55},
|
||||
{"datum": "15.09.", "puls": 53},
|
||||
{"datum": "16.09.", "puls": 51}
|
||||
]
|
||||
EOF
|
||||
```
|
||||
|
||||
### 3. Donut-Chart für Anteile / Verteilungen
|
||||
|
||||
```bash
|
||||
uv run /home/node/mop-skills/mop-skills/skills/chart-to-chat/scripts/chart.py \
|
||||
--title "Ausgaben September" \
|
||||
--type donut \
|
||||
--data '[{"kategorie": "Miete", "euro": 950}, {"kategorie": "Lebensmittel", "euro": 420}, {"kategorie": "Sport", "euro": 110}]'
|
||||
```
|
||||
|
||||
### 4. Optionen des CLI-Tools
|
||||
|
||||
- `--title "<Text>"`: Überschrift des Diagramms.
|
||||
- `--notes "<Text>"`: Optionale Unterzeile / Erläuterung.
|
||||
- `--type bar|hbar|line|area|donut|kpi|table`: Visualisierungstyp.
|
||||
- `--data "<JSON>"` oder `--data -` (Stdin): Eingabedaten.
|
||||
- `--x <Spaltenname>` / `--y <Spaltenname>`: Manuelle Angabe der Achsenspalten (wird ansonsten automatisch erkannt).
|
||||
- `--engine auto|dct|quickchart`: `auto` nutzt primär das lokale `dbt-charts` und fällt bei Bedarf transparent auf `QuickChart.io` zurück.
|
||||
|
||||
---
|
||||
|
||||
## 📱 Auslieferung an Telegram
|
||||
|
||||
Das Skript gibt als letzte Zeile den Pfad der erzeugten PNG-Datei im Format `MEDIA:<pfad>` aus, z. B.:
|
||||
```text
|
||||
MEDIA:/tmp/charts/chart_a1b2c3d4.png
|
||||
```
|
||||
|
||||
Schließe diese Zeile einfach in deine finale Antwort an den Nutzer ein. OpenClaw versendet das Diagramm dadurch automatisch als Bild im Chat:
|
||||
|
||||
```markdown
|
||||
Hier ist die Übersicht deiner Laufdistanzen für diese Woche:
|
||||
|
||||
MEDIA:/tmp/charts/chart_a1b2c3d4.png
|
||||
```
|
||||
Executable
+456
@@ -0,0 +1,456 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
chart.py — Universal Chart & Graph Generator for OpenClaw / Telegram
|
||||
|
||||
Generates clean, publication-grade chart images from JSON/CSV data using:
|
||||
1. Primary: dbt-charts (`dct render`) — local, offline, Vega-Lite/resvg based.
|
||||
2. Fallback: QuickChart.io — fast HTTP API (Chart.js), zero local dependencies.
|
||||
|
||||
Outputs `MEDIA:<path>` on stdout for native Telegram image delivery.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional, Tuple, Union
|
||||
|
||||
|
||||
KNOWN_DCT_PATHS = [
|
||||
Path("/home/node/.openclaw/.local/bin/dct"),
|
||||
Path("/home/openclaw/.local/bin/dct"),
|
||||
Path("/home/node/.openclaw/.local/share/uv/tools/dbt-charts/bin/dct"),
|
||||
Path("/home/openclaw/.local/share/uv/tools/dbt-charts/bin/dct"),
|
||||
Path.home() / ".local/bin/dct",
|
||||
]
|
||||
|
||||
|
||||
def find_dct_executable() -> Optional[str]:
|
||||
"""Locate the dct executable in PATH or well-known install locations."""
|
||||
# 1. System PATH
|
||||
found = shutil.which("dct")
|
||||
if found:
|
||||
return found
|
||||
|
||||
# 2. Well-known absolute container/local paths
|
||||
for candidate in KNOWN_DCT_PATHS:
|
||||
if candidate.is_file() and os.access(candidate, os.X_OK):
|
||||
return str(candidate)
|
||||
|
||||
# 3. Check if uv tool run dct works
|
||||
uv_path = shutil.which("uv") or "/home/node/.openclaw/bin/uv"
|
||||
if os.path.exists(uv_path) and os.access(uv_path, os.X_OK):
|
||||
try:
|
||||
res = subprocess.run(
|
||||
[uv_path, "tool", "run", "dct", "--version"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
if res.returncode == 0:
|
||||
return "uv_tool_run"
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def parse_input_data(raw_data: str) -> Tuple[List[str], List[List[Any]], Optional[str], Optional[str]]:
|
||||
"""
|
||||
Parse input string (JSON or CSV) into (columns, rows, suggested_x, suggested_y).
|
||||
Supports:
|
||||
- List of dicts: [{"label": "A", "val": 10}, {"label": "B", "val": 20}]
|
||||
- Dict of values: {"A": 10, "B": 20}
|
||||
- Explicit table: {"columns": ["k", "v"], "values": [["A", 10], ["B", 20]]}
|
||||
- Single KPI: {"value": 1284, "label": "Revenue", "delta": "+12%"}
|
||||
"""
|
||||
raw_data = raw_data.strip()
|
||||
if not raw_data:
|
||||
raise ValueError("Input data is empty.")
|
||||
|
||||
# Try JSON
|
||||
parsed = json.loads(raw_data)
|
||||
|
||||
# Case 1: Dict with explicit columns & values
|
||||
if isinstance(parsed, dict) and "columns" in parsed and ("values" in parsed or "rows" in parsed):
|
||||
cols = [str(c) for c in parsed["columns"]]
|
||||
rows = parsed.get("values") or parsed.get("rows") or []
|
||||
x_col = cols[0] if len(cols) > 0 else None
|
||||
y_col = cols[1] if len(cols) > 1 else None
|
||||
return cols, rows, x_col, y_col
|
||||
|
||||
# Case 2: Dict of key-value pairs (e.g. {"Mo": 10, "Di": 15})
|
||||
if isinstance(parsed, dict) and not any(k in parsed for k in ["columns", "values", "rows", "type", "chart"]):
|
||||
cols = ["category", "value"]
|
||||
rows = [[k, v] for k, v in parsed.items()]
|
||||
return cols, rows, "category", "value"
|
||||
|
||||
# Case 3: List of dicts (most common agent format)
|
||||
if isinstance(parsed, list):
|
||||
if not parsed:
|
||||
raise ValueError("Data list is empty.")
|
||||
if isinstance(parsed[0], dict):
|
||||
# Collect unique keys maintaining insertion order
|
||||
cols: List[str] = []
|
||||
for item in parsed:
|
||||
for k in item.keys():
|
||||
if k not in cols:
|
||||
cols.append(k)
|
||||
|
||||
rows: List[List[Any]] = []
|
||||
for item in parsed:
|
||||
rows.append([item.get(c) for c in cols])
|
||||
|
||||
# Infer x (category/date) and y (measure)
|
||||
x_col = None
|
||||
y_col = None
|
||||
for c in cols:
|
||||
low = c.lower()
|
||||
if low in ["x", "label", "name", "date", "datum", "month", "monat", "day", "tag", "category", "kategorie", "typ", "woche", "week"]:
|
||||
x_col = c
|
||||
break
|
||||
if not x_col and len(cols) > 0:
|
||||
x_col = cols[0]
|
||||
|
||||
for c in cols:
|
||||
if c == x_col:
|
||||
continue
|
||||
low = c.lower()
|
||||
if low in ["y", "value", "wert", "count", "anzahl", "amount", "menge", "total", "summe", "preis", "price", "revenue", "umsatz", "km", "kmh", "watt", "hr", "puls"]:
|
||||
y_col = c
|
||||
break
|
||||
if not y_col and len(cols) > 1:
|
||||
y_col = cols[1]
|
||||
|
||||
return cols, rows, x_col, y_col
|
||||
|
||||
elif isinstance(parsed[0], list):
|
||||
# List of lists without explicit headers
|
||||
cols = [f"col_{i+1}" for i in range(len(parsed[0]))]
|
||||
return cols, parsed, cols[0], cols[1] if len(cols) > 1 else None
|
||||
|
||||
raise ValueError(f"Unrecognized data structure: {type(parsed)}")
|
||||
|
||||
|
||||
def render_dbt_chart(
|
||||
dct_cmd: str,
|
||||
title: str,
|
||||
chart_type: str,
|
||||
columns: List[str],
|
||||
rows: List[List[Any]],
|
||||
x_col: str,
|
||||
y_col: str,
|
||||
output_path: Path,
|
||||
notes: Optional[str] = None,
|
||||
theme: Optional[str] = None,
|
||||
) -> bool:
|
||||
"""Render a chart using dbt-charts (dct render)."""
|
||||
scratch_dir = Path(tempfile.mkdtemp(prefix="dct_render_"))
|
||||
try:
|
||||
# 1. dbt_charts.yml project file (mandatory for dct)
|
||||
(scratch_dir / "dbt_charts.yml").write_text("# dbt charts project\nversion: 1\n", encoding="utf-8")
|
||||
|
||||
# Normalize chart type
|
||||
ct = chart_type.lower()
|
||||
if ct in ["hbar", "horizontal_bar"]:
|
||||
ct = "bar"
|
||||
is_horizontal = True
|
||||
else:
|
||||
is_horizontal = False
|
||||
|
||||
if ct in ["donut", "pie"]:
|
||||
dbt_type = "donut"
|
||||
elif ct in ["line", "trend"]:
|
||||
dbt_type = "line"
|
||||
elif ct in ["area"]:
|
||||
dbt_type = "area"
|
||||
elif ct in ["kpi"]:
|
||||
dbt_type = "kpi"
|
||||
elif ct in ["table"]:
|
||||
dbt_type = "table"
|
||||
else:
|
||||
dbt_type = "bar"
|
||||
|
||||
# Build board YAML
|
||||
board: Dict[str, Any] = {
|
||||
"title": title or "Dashboard",
|
||||
"queries": {
|
||||
"dataset": {
|
||||
"columns": columns,
|
||||
"values": rows,
|
||||
}
|
||||
},
|
||||
"charts": {},
|
||||
"rows": ["main_chart"],
|
||||
}
|
||||
|
||||
if notes:
|
||||
board["notes"] = notes
|
||||
|
||||
chart_spec: Dict[str, Any] = {
|
||||
"type": dbt_type,
|
||||
"query": "dataset",
|
||||
}
|
||||
if title:
|
||||
chart_spec["title"] = title
|
||||
|
||||
if dbt_type in ["donut", "pie"]:
|
||||
chart_spec["theta"] = y_col
|
||||
chart_spec["color"] = x_col
|
||||
elif dbt_type == "kpi":
|
||||
chart_spec["value"] = y_col or columns[0]
|
||||
chart_spec["label"] = x_col or "Value"
|
||||
elif dbt_type == "table":
|
||||
# Table automatically uses query columns
|
||||
pass
|
||||
else:
|
||||
# Bar / Line / Area
|
||||
if is_horizontal:
|
||||
chart_spec["x"] = y_col
|
||||
chart_spec["y"] = x_col
|
||||
else:
|
||||
chart_spec["x"] = x_col
|
||||
chart_spec["y"] = y_col
|
||||
|
||||
board["charts"]["main_chart"] = chart_spec
|
||||
|
||||
import yaml # PyYAML is standard in python / uv environments
|
||||
board_file = scratch_dir / "board.yml"
|
||||
board_file.write_text(yaml.safe_dump(board, sort_keys=False), encoding="utf-8")
|
||||
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Build command
|
||||
if dct_cmd == "uv_tool_run":
|
||||
cmd = ["uv", "tool", "run", "dct", "render", str(board_file), "--format", "png", "--output", str(output_path)]
|
||||
else:
|
||||
cmd = [dct_cmd, "render", str(board_file), "--format", "png", "--output", str(output_path)]
|
||||
|
||||
proc = subprocess.run(
|
||||
cmd,
|
||||
cwd=str(scratch_dir),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
if proc.returncode == 0 and output_path.exists() and output_path.stat().st_size > 0:
|
||||
return True
|
||||
else:
|
||||
sys.stderr.write(f"[dct render warning] exit={proc.returncode} stderr={proc.stderr.strip()}\n")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
sys.stderr.write(f"[dct render error] {e}\n")
|
||||
return False
|
||||
finally:
|
||||
shutil.rmtree(scratch_dir, ignore_errors=True)
|
||||
|
||||
|
||||
def render_quickchart(
|
||||
title: str,
|
||||
chart_type: str,
|
||||
columns: List[str],
|
||||
rows: List[List[Any]],
|
||||
x_col: str,
|
||||
y_col: str,
|
||||
output_path: Path,
|
||||
width: int = 800,
|
||||
height: int = 450,
|
||||
) -> bool:
|
||||
"""Render chart via QuickChart.io API as fallback."""
|
||||
try:
|
||||
x_idx = columns.index(x_col)
|
||||
y_idx = columns.index(y_col)
|
||||
|
||||
labels = [str(r[x_idx]) for r in rows]
|
||||
values = []
|
||||
for r in rows:
|
||||
v = r[y_idx]
|
||||
try:
|
||||
values.append(float(v) if v is not None else 0)
|
||||
except (ValueError, TypeError):
|
||||
values.append(0)
|
||||
|
||||
ct = chart_type.lower()
|
||||
if ct in ["donut", "pie"]:
|
||||
qc_type = "doughnut" if ct == "donut" else "pie"
|
||||
bg_colors = [
|
||||
"#3366cc", "#dc3912", "#ff9900", "#109618", "#990099",
|
||||
"#3b3eac", "#0099c6", "#dd4477", "#66aa00", "#b82e2e",
|
||||
]
|
||||
dataset = {
|
||||
"data": values,
|
||||
"backgroundColor": bg_colors[:len(values)],
|
||||
}
|
||||
elif ct in ["line", "trend"]:
|
||||
qc_type = "line"
|
||||
dataset = {
|
||||
"label": y_col,
|
||||
"data": values,
|
||||
"borderColor": "#3e64ff",
|
||||
"backgroundColor": "rgba(62, 100, 255, 0.1)",
|
||||
"fill": False,
|
||||
"tension": 0.2,
|
||||
}
|
||||
elif ct in ["area"]:
|
||||
qc_type = "line"
|
||||
dataset = {
|
||||
"label": y_col,
|
||||
"data": values,
|
||||
"borderColor": "#3e64ff",
|
||||
"backgroundColor": "rgba(62, 100, 255, 0.25)",
|
||||
"fill": True,
|
||||
"tension": 0.2,
|
||||
}
|
||||
else:
|
||||
qc_type = "bar"
|
||||
dataset = {
|
||||
"label": y_col,
|
||||
"data": values,
|
||||
"backgroundColor": "#3e64ff",
|
||||
}
|
||||
|
||||
chart_config = {
|
||||
"type": qc_type,
|
||||
"data": {
|
||||
"labels": labels,
|
||||
"datasets": [dataset],
|
||||
},
|
||||
"options": {
|
||||
"title": {
|
||||
"display": bool(title),
|
||||
"text": title,
|
||||
"fontSize": 18,
|
||||
},
|
||||
"legend": {
|
||||
"display": qc_type in ["pie", "doughnut"],
|
||||
"position": "top",
|
||||
},
|
||||
"plugins": {
|
||||
"datalabels": {
|
||||
"display": False,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
query_params = {
|
||||
"w": str(width),
|
||||
"h": str(height),
|
||||
"devicePixelRatio": "2.0",
|
||||
"c": json.dumps(chart_config),
|
||||
}
|
||||
url = "https://quickchart.io/chart?" + urllib.parse.urlencode(query_params)
|
||||
|
||||
req = urllib.request.Request(url, headers={"User-Agent": "OpenClaw-Chart/1.0"})
|
||||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||||
if resp.status == 200:
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_bytes(resp.read())
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
sys.stderr.write(f"[quickchart error] {e}\n")
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="OpenClaw Universal Chart Generator")
|
||||
parser.add_argument("--data", help="JSON data string, or '-' for stdin")
|
||||
parser.add_argument("--file", help="Path to JSON or CSV file")
|
||||
parser.add_argument("--type", default="bar", choices=["bar", "hbar", "line", "area", "donut", "pie", "kpi", "table"], help="Chart type")
|
||||
parser.add_argument("--title", default="Übersicht", help="Chart or Dashboard title")
|
||||
parser.add_argument("--notes", help="Subtitle or explanatory notes")
|
||||
parser.add_argument("--x", help="X-axis / category column name")
|
||||
parser.add_argument("--y", help="Y-axis / measure column name")
|
||||
parser.add_argument("--output", help="Target PNG path (default: /tmp/charts/chart_<hash>.png)")
|
||||
parser.add_argument("--engine", default="auto", choices=["auto", "dct", "quickchart"], help="Rendering engine")
|
||||
parser.add_argument("--width", type=int, default=800, help="Width in pixels (QuickChart)")
|
||||
parser.add_argument("--height", type=int, default=450, help="Height in pixels (QuickChart)")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Read data
|
||||
raw_data = None
|
||||
if args.data and args.data != "-":
|
||||
raw_data = args.data
|
||||
elif args.file:
|
||||
raw_data = Path(args.file).read_text(encoding="utf-8")
|
||||
elif args.data == "-" or not sys.stdin.isatty():
|
||||
raw_data = sys.stdin.read()
|
||||
|
||||
if not raw_data:
|
||||
parser.error("No input data provided. Use --data, --file, or pipe data via STDIN.")
|
||||
|
||||
try:
|
||||
columns, rows, inferred_x, inferred_y = parse_input_data(raw_data)
|
||||
except Exception as e:
|
||||
sys.stderr.write(f"Error parsing input data: {e}\n")
|
||||
sys.exit(1)
|
||||
|
||||
x_col = args.x or inferred_x or (columns[0] if columns else "x")
|
||||
y_col = args.y or inferred_y or (columns[1] if len(columns) > 1 else columns[0])
|
||||
|
||||
# Destination path
|
||||
if args.output:
|
||||
output_path = Path(args.output)
|
||||
else:
|
||||
content_hash = hashlib.md5(f"{args.title}_{args.type}_{raw_data}".encode("utf-8")).hexdigest()[:8]
|
||||
output_path = Path(f"/tmp/charts/chart_{content_hash}.png")
|
||||
|
||||
rendered = False
|
||||
|
||||
# Attempt 1: dbt Charts (unless engine is explicitly quickchart)
|
||||
if args.engine in ["auto", "dct"]:
|
||||
dct_bin = find_dct_executable()
|
||||
if dct_bin:
|
||||
rendered = render_dbt_chart(
|
||||
dct_cmd=dct_bin,
|
||||
title=args.title,
|
||||
chart_type=args.type,
|
||||
columns=columns,
|
||||
rows=rows,
|
||||
x_col=x_col,
|
||||
y_col=y_col,
|
||||
output_path=output_path,
|
||||
notes=args.notes,
|
||||
)
|
||||
elif args.engine == "dct":
|
||||
sys.stderr.write("Error: dct executable not found and --engine dct was requested.\n")
|
||||
sys.exit(1)
|
||||
|
||||
# Attempt 2: QuickChart Fallback
|
||||
if not rendered and args.engine in ["auto", "quickchart"]:
|
||||
rendered = render_quickchart(
|
||||
title=args.title,
|
||||
chart_type=args.type,
|
||||
columns=columns,
|
||||
rows=rows,
|
||||
x_col=x_col,
|
||||
y_col=y_col,
|
||||
output_path=output_path,
|
||||
width=args.width,
|
||||
height=args.height,
|
||||
)
|
||||
|
||||
if not rendered:
|
||||
sys.stderr.write("Error: Chart rendering failed on all available engines.\n")
|
||||
sys.exit(1)
|
||||
|
||||
# Deliver MEDIA line for Telegram/OpenClaw
|
||||
print(f"MEDIA:{output_path.resolve()}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,133 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Unit and integration tests for chart.py
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
# Add scripts directory to path
|
||||
SCRIPTS_DIR = Path(__file__).resolve().parent
|
||||
sys.path.insert(0, str(SCRIPTS_DIR))
|
||||
|
||||
from chart import parse_input_data
|
||||
|
||||
|
||||
class TestChartParser(unittest.TestCase):
|
||||
def test_list_of_dicts(self):
|
||||
raw = json.dumps([
|
||||
{"day": "Mo", "temp": 18},
|
||||
{"day": "Di", "temp": 22},
|
||||
{"day": "Mi", "temp": 19},
|
||||
])
|
||||
cols, rows, x, y = parse_input_data(raw)
|
||||
self.assertEqual(cols, ["day", "temp"])
|
||||
self.assertEqual(len(rows), 3)
|
||||
self.assertEqual(x, "day")
|
||||
self.assertEqual(y, "temp")
|
||||
|
||||
def test_dict_key_value(self):
|
||||
raw = json.dumps({"Bier": 4.5, "Pizza": 12.0, "Kaffee": 3.2})
|
||||
cols, rows, x, y = parse_input_data(raw)
|
||||
self.assertEqual(cols, ["category", "value"])
|
||||
self.assertEqual(len(rows), 3)
|
||||
self.assertEqual(x, "category")
|
||||
self.assertEqual(y, "value")
|
||||
|
||||
def test_explicit_columns_values(self):
|
||||
raw = json.dumps({
|
||||
"columns": ["Monat", "Umsatz"],
|
||||
"values": [["Jan", 100], ["Feb", 150]]
|
||||
})
|
||||
cols, rows, x, y = parse_input_data(raw)
|
||||
self.assertEqual(cols, ["Monat", "Umsatz"])
|
||||
self.assertEqual(len(rows), 2)
|
||||
self.assertEqual(x, "Monat")
|
||||
self.assertEqual(y, "Umsatz")
|
||||
|
||||
|
||||
class TestChartCLI(unittest.TestCase):
|
||||
def run_cli(self, args, stdin_input=None):
|
||||
cmd = [sys.executable, str(SCRIPTS_DIR / "chart.py")] + args
|
||||
res = subprocess.run(
|
||||
cmd,
|
||||
input=stdin_input,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
return res
|
||||
|
||||
def test_bar_chart_generation(self):
|
||||
data = json.dumps([
|
||||
{"label": "Projekt A", "stunden": 14},
|
||||
{"label": "Projekt B", "stunden": 28},
|
||||
{"label": "Projekt C", "stunden": 8},
|
||||
])
|
||||
out_path = Path("/tmp/charts/test_bar.png")
|
||||
if out_path.exists():
|
||||
out_path.unlink()
|
||||
|
||||
res = self.run_cli(["--type", "bar", "--title", "Projektstunden", "--data", data, "--output", str(out_path)])
|
||||
self.assertEqual(res.returncode, 0, f"Stderr: {res.stderr}")
|
||||
self.assertTrue(res.stdout.strip().startswith("MEDIA:"))
|
||||
self.assertTrue(out_path.exists())
|
||||
self.assertGreater(out_path.stat().st_size, 1000)
|
||||
|
||||
def test_line_chart_stdin(self):
|
||||
data = json.dumps([
|
||||
{"datum": "2026-09-10", "puls": 62},
|
||||
{"datum": "2026-09-11", "puls": 58},
|
||||
{"datum": "2026-09-12", "puls": 65},
|
||||
{"datum": "2026-09-13", "puls": 59},
|
||||
])
|
||||
out_path = Path("/tmp/charts/test_line.png")
|
||||
if out_path.exists():
|
||||
out_path.unlink()
|
||||
|
||||
res = self.run_cli(["--type", "line", "--title", "Ruhepuls Verlauf", "--data", "-", "--output", str(out_path)], stdin_input=data)
|
||||
self.assertEqual(res.returncode, 0, f"Stderr: {res.stderr}")
|
||||
self.assertTrue(res.stdout.strip().startswith("MEDIA:"))
|
||||
self.assertTrue(out_path.exists())
|
||||
self.assertGreater(out_path.stat().st_size, 1000)
|
||||
|
||||
def test_donut_chart(self):
|
||||
data = json.dumps([
|
||||
{"bereich": "Wohnen", "kosten": 950},
|
||||
{"bereich": "Essen", "kosten": 450},
|
||||
{"bereich": "Mobilität", "kosten": 120},
|
||||
{"bereich": "Freizeit", "kosten": 200},
|
||||
])
|
||||
out_path = Path("/tmp/charts/test_donut.png")
|
||||
if out_path.exists():
|
||||
out_path.unlink()
|
||||
|
||||
res = self.run_cli(["--type", "donut", "--title", "Ausgaben", "--data", data, "--output", str(out_path)])
|
||||
self.assertEqual(res.returncode, 0, f"Stderr: {res.stderr}")
|
||||
self.assertTrue(res.stdout.strip().startswith("MEDIA:"))
|
||||
self.assertTrue(out_path.exists())
|
||||
self.assertGreater(out_path.stat().st_size, 1000)
|
||||
|
||||
def test_quickchart_fallback_engine(self):
|
||||
data = json.dumps([
|
||||
{"fach": "Mathe", "note": 1.7},
|
||||
{"fach": "Physik", "note": 2.0},
|
||||
{"fach": "Informatik", "note": 1.0},
|
||||
])
|
||||
out_path = Path("/tmp/charts/test_quickchart.png")
|
||||
if out_path.exists():
|
||||
out_path.unlink()
|
||||
|
||||
res = self.run_cli(["--engine", "quickchart", "--type", "bar", "--title", "Notenspiegel", "--data", data, "--output", str(out_path)])
|
||||
self.assertEqual(res.returncode, 0, f"Stderr: {res.stderr}")
|
||||
self.assertTrue(res.stdout.strip().startswith("MEDIA:"))
|
||||
self.assertTrue(out_path.exists())
|
||||
self.assertGreater(out_path.stat().st_size, 1000)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user