"""render_chart — declarative chart-spec builder (SPINE_V2_PLAN §4.1, S2). First tool of the `render_*` family: turns an already-materialized DataFrame into a Plotly-conformant JSON spec wrapped in the `dataeyond.chart.v1` envelope (SPINE_V2_PLAN §4.2). The FE renders it with plotly.js (`Plotly.newPlot(el, spec.plotly.data, spec.plotly.layout)`). Deliberately NOT a code generator and NOT a plotly import (locked decision, DEV_PLAN deferred row #26): the spec is a hand-built dict, so no generated code ever executes and no new dependency lands. Style (colors, axis chrome) is the fixed module preset below — never a planner/LLM decision (CoDA's "Design Explorer" phase collapsed to a constant). Pattern A, same as the `analyze_*` family: `data` is resolved to a DataFrame by the invoker from an upstream table-kind output; this function never fetches. """ from __future__ import annotations import datetime as _dt from decimal import Decimal import numpy as np import pandas as pd from src.tools.analytics.descriptive import ColumnNotFoundError # v1 chart types (SPINE_V2_PLAN §4.1). `pie` maps x -> labels, y -> values. CHART_TYPES = ("bar", "line", "pie", "scatter") class UnsupportedChartTypeError(ValueError): """The requested chart_type is not in CHART_TYPES (error_code UNSUPPORTED_CHART_TYPE).""" # --------------------------------------------------------------------------- # # House style preset — fixed by design, not a planner argument. # # Colorway = the dataviz reference categorical palette (8 slots, light mode). # The slot ORDER is the colorblind-safety mechanism (validated: worst adjacent # CVD deltaE 24.2 vs the >=12 target) — series take slots in this fixed order, # never cycled or re-picked. Backgrounds are transparent so the chart inherits # the FE surface; inks/gridlines are the palette's chrome roles. # --------------------------------------------------------------------------- # COLORWAY = [ "#2a78d6", # blue "#1baf7a", # aqua "#eda100", # yellow "#008300", # green "#4a3aa7", # violet "#e34948", # red "#e87ba4", # magenta "#eb6834", # orange ] _INK_SECONDARY = "#52514e" # body text _INK_MUTED = "#898781" # axis titles / tick labels (mode-neutral) _GRIDLINE = "#e1e0d9" # hairline grid _BASELINE = "#c3c2b7" # axis line LAYOUT_PRESET: dict[str, object] = { "colorway": COLORWAY, "font": { "family": 'system-ui, -apple-system, "Segoe UI", sans-serif', "size": 13, "color": _INK_SECONDARY, }, "paper_bgcolor": "rgba(0,0,0,0)", "plot_bgcolor": "rgba(0,0,0,0)", "margin": {"t": 48, "r": 16, "b": 48, "l": 56}, "bargap": 0.25, # keeps adjacent bars visually separated (mark-spec gap) } # Mark specs: thin marks — 2px lines, 8px markers. _LINE_STYLE = {"width": 2} _MARKER_STYLE = {"size": 8} def _axis(label: str) -> dict[str, object]: """Recessive axis chrome + the column name as the axis title.""" return { "title": {"text": label, "font": {"color": _INK_MUTED}}, "tickfont": {"color": _INK_MUTED}, "gridcolor": _GRIDLINE, "linecolor": _BASELINE, "automargin": True, } def _clean(value: object) -> object: """One JSON-safe scalar: numpy -> Python, NaN/NaT -> None, dates -> ISO strings. Chart specs are persisted as JSONB and shipped to the FE, so every value in the trace arrays must survive json.dumps. Upstream `_materialize` already normalizes fully-numeric columns, but mixed columns can still carry Decimal, and date columns carry Timestamps. """ if value is None or value is pd.NaT: return None if isinstance(value, float) and np.isnan(value): return None if hasattr(value, "item"): # numpy scalar (incl. np.datetime64 via Timestamp below) value = value.item() if isinstance(value, float) and np.isnan(value): return None if isinstance(value, Decimal): return float(value) if isinstance(value, pd.Timestamp): return value.isoformat() if isinstance(value, _dt.datetime | _dt.date): return value.isoformat() if isinstance(value, bool | int | float | str): return value return str(value) def _values(col: pd.Series) -> list[object]: return [_clean(v) for v in col.tolist()] def _series_label(value: object) -> str: """Trace name for one series group; null groups get an explicit label.""" cleaned = _clean(value) return "(missing)" if cleaned is None else str(cleaned) # Prompt-style description read by the Planner to decide WHEN to pick this tool. DESCRIPTION = """\ Summary: Build a chart specification (bar, line, pie, scatter) from an \ already-retrieved table, for the app to render. Maps one x column and one y \ column (pie: x -> slice labels, y -> slice values); an optional `series` column \ splits bar/line/scatter into one trace per category. Produces a spec only — it \ computes no numbers. USE WHEN the user EXPLICITLY asks to see a chart — "plot", "chart", "graph", \ "visualize", "diagram" (ID: "grafik", "diagram", "plot", "visualisasikan", \ "buatkan grafik/diagram"). ALWAYS a tail step: `data` consumes the output of an \ upstream task that yields a table (retrieve_data or a table-kind analyze_*). DON'T USE WHEN: - the user did not explicitly ask for a chart -> answer with tables/stats only \ (never add a speculative chart) - the numbers still need computing -> run the analyze_*/retrieve step first, \ then chart its table output - the upstream output is stats- or series-kind -> feed it a table-kind task \ (e.g. a grouped retrieve or analyze_aggregate) Example questions: - "plot revenue by region as a bar chart" - "buatkan grafik penjualan per kategori" - "show a pie chart of orders by sales channel" - "visualize monthly revenue" (chain a month-grouped table first, then line) """ def render_chart( df: pd.DataFrame, chart_type: str, x: str, y: str, series: str | None = None, title: str | None = None, ) -> dict[str, object]: """Build a `dataeyond.chart.v1` envelope from a table (SPINE_V2_PLAN §4.2). Args: df: already-materialized data (the invoker resolves `data` upstream). x: column for the x axis (pie: slice labels). y: column for the y axis (pie: slice values). series: optional column splitting bar/line/scatter into one trace per distinct value (fixed-order colorway slots). Ignored for pie. title: chart title; defaults to " by ". Returns: The envelope dict: {schema, chart_type, title, plotly: {data, layout}}. Raises: UnsupportedChartTypeError: chart_type not in CHART_TYPES. ColumnNotFoundError: x, y, or series absent from the data. """ if chart_type not in CHART_TYPES: raise UnsupportedChartTypeError( f"unsupported chart_type '{chart_type}'; supported: {list(CHART_TYPES)}" ) needed = [x, y] if chart_type == "pie" or series is None else [x, y, series] missing = [c for c in needed if c not in df.columns] if missing: raise ColumnNotFoundError(f"columns not found: {missing}") chart_title = title or f"{y} by {x}" if chart_type == "pie": traces: list[dict[str, object]] = [ {"type": "pie", "labels": _values(df[x]), "values": _values(df[y])} ] else: plotly_type = "bar" if chart_type == "bar" else "scatter" mode = {"line": "lines+markers", "scatter": "markers"}.get(chart_type) def _trace(frame: pd.DataFrame, name: str) -> dict[str, object]: t: dict[str, object] = { "type": plotly_type, "x": _values(frame[x]), "y": _values(frame[y]), "name": name, } if mode is not None: t["mode"] = mode t["line"] = dict(_LINE_STYLE) t["marker"] = dict(_MARKER_STYLE) return t if series is None: traces = [_trace(df, y)] else: # sort=True keeps trace order (and therefore colorway slot # assignment) deterministic across runs of the same data. traces = [ _trace(group, _series_label(value)) for value, group in df.groupby(series, dropna=False, sort=True) ] layout: dict[str, object] = { **LAYOUT_PRESET, "title": {"text": chart_title}, # Legend only when there is more than one thing to identify: multi-series # traces, or pie slices (a single named series is titled, not legended). "showlegend": len(traces) > 1 or chart_type == "pie", } if chart_type != "pie": layout["xaxis"] = _axis(x) layout["yaxis"] = _axis(y) return { "schema": "dataeyond.chart.v1", "chart_type": chart_type, "title": chart_title, "plotly": {"data": traces, "layout": layout}, }