| 1 | """Extract native ChartEx payloads from PPTX chart parts. |
| 2 | |
| 3 | The parser is deliberately closed around the seven ChartEx data models that |
| 4 | the native writer already emits. Data topology and caches must be complete; |
| 5 | unmodeled chart chrome, labels, axes, binning options, and style details are |
| 6 | allowed to normalize during editable native reconstruction. |
| 7 | """ |
| 8 | |
| 9 | from __future__ import annotations |
| 10 | |
| 11 | import math |
| 12 | import re |
| 13 | from typing import Any |
| 14 | from xml.etree import ElementTree as ET |
| 15 | |
| 16 | from svg_to_pptx.native_objects.chart_data import validate_chart_payload |
| 17 | from svg_to_pptx.native_objects.marker_common import CHART_COLOR_STYLE_REL_TYPE |
| 18 | |
| 19 | from .color_resolver import COLOR_TAGS, ColorPalette, resolve_color |
| 20 | from .emu_units import NS, Xfrm |
| 21 | from .ooxml_loader import OoxmlPackage, PartRef |
| 22 | |
| 23 | |
| 24 | CHARTEX_URI = "http://schemas.microsoft.com/office/drawing/2014/chartex" |
| 25 | |
| 26 | CX_NS = { |
| 27 | **NS, |
| 28 | "cx": CHARTEX_URI, |
| 29 | } |
| 30 | |
| 31 | _INT_TOKEN_RE = re.compile(r"[0-9]+") |
| 32 | _HIERARCHY_LAYOUTS = {"sunburst", "treemap"} |
| 33 | _FLAT_LAYOUTS = {"funnel", "waterfall"} |
| 34 | _LEGEND_POSITIONS = {"b", "l", "r", "t"} |
| 35 | |
| 36 | |
| 37 | class UnsupportedChartEx(RuntimeError): |
| 38 | """A ChartEx part cannot map to the current native authoring schema.""" |
| 39 | |
| 40 | def __init__(self, status: str) -> None: |
| 41 | super().__init__(status) |
| 42 | self.status = status |
| 43 | |
| 44 | |
| 45 | def extract_native_chartex_payload( |
| 46 | graphic_data: ET.Element | None, |
| 47 | xfrm: Xfrm, |
| 48 | slide_part: PartRef, |
| 49 | pkg: OoxmlPackage, |
| 50 | palette: ColorPalette | None = None, |
| 51 | ) -> dict[str, Any]: |
| 52 | """Return a writer-valid payload for one supported ChartEx reference. |
| 53 | |
| 54 | ``UnsupportedChartEx.status`` is suitable for |
| 55 | ``data-pptx-replacement-status`` when the reference, relationship, data |
| 56 | topology, or cache is unusable. |
| 57 | Style-part failures never reject an otherwise valid data payload. |
| 58 | """ |
| 59 | if graphic_data is None: |
| 60 | raise UnsupportedChartEx("unsupported-chart-reference") |
| 61 | if xfrm.rot or xfrm.flip_h or xfrm.flip_v: |
| 62 | raise UnsupportedChartEx("unsupported-native-transform") |
| 63 | |
| 64 | chart_ref = graphic_data.find("cx:chart", CX_NS) |
| 65 | if chart_ref is None: |
| 66 | raise UnsupportedChartEx("unsupported-chart-reference") |
| 67 | rid = chart_ref.attrib.get(f"{{{NS['r']}}}id") |
| 68 | if not rid: |
| 69 | raise UnsupportedChartEx("unsupported-chart-reference") |
| 70 | |
| 71 | chart_path = slide_part.resolve_rel(rid) |
| 72 | if not chart_path: |
| 73 | raise UnsupportedChartEx("unsupported-chart-relationship") |
| 74 | chart_part = pkg.load_part(chart_path) |
| 75 | if chart_part is None: |
| 76 | raise UnsupportedChartEx("unsupported-chart-part") |
| 77 | |
| 78 | try: |
| 79 | payload = _payload_from_chartex_xml(chart_part.xml, xfrm) |
| 80 | colors = _resolved_chart_colors(chart_part, pkg, palette) |
| 81 | if colors: |
| 82 | payload["style"] = {"colors": colors} |
| 83 | validate_chart_payload(payload) |
| 84 | except UnsupportedChartEx: |
| 85 | raise |
| 86 | except RuntimeError as exc: |
| 87 | raise UnsupportedChartEx("unsupported-chartex-schema") from exc |
| 88 | except (AttributeError, OverflowError, TypeError, ValueError) as exc: |
| 89 | raise UnsupportedChartEx("unsupported-chartex-parse") from exc |
| 90 | return payload |
| 91 | |
| 92 | |
| 93 | def _payload_from_chartex_xml( |
| 94 | chart_root: ET.Element, |
| 95 | xfrm: Xfrm, |
| 96 | ) -> dict[str, Any]: |
| 97 | if chart_root.tag != f"{{{CHARTEX_URI}}}chartSpace": |
| 98 | raise UnsupportedChartEx("unsupported-chartex-part") |
| 99 | |
| 100 | chart_data = _one_child(chart_root, "chartData", "unsupported-chartex-structure") |
| 101 | chart = _one_child(chart_root, "chart", "unsupported-chartex-structure") |
| 102 | plot_area = _one_child(chart, "plotArea", "unsupported-chartex-structure") |
| 103 | region = _one_child(plot_area, "plotAreaRegion", "unsupported-chartex-structure") |
| 104 | |
| 105 | data_by_id = _data_parts(chart_data) |
| 106 | series_nodes = _children(region, "series") |
| 107 | if not series_nodes: |
| 108 | raise UnsupportedChartEx("unsupported-chartex-series") |
| 109 | |
| 110 | chart_type = _chart_type(series_nodes) |
| 111 | payload: dict[str, Any] = { |
| 112 | **_bounds_payload(xfrm), |
| 113 | "type": chart_type, |
| 114 | } |
| 115 | |
| 116 | referenced_data_ids: list[int] |
| 117 | if chart_type in _HIERARCHY_LAYOUTS: |
| 118 | data_id = _series_data_id(series_nodes[0]) |
| 119 | data = _require_data(data_by_id, data_id) |
| 120 | raw_levels = _string_dimension(data, expected_levels=None) |
| 121 | values = _numeric_dimension(data, "size") |
| 122 | if not raw_levels or any(len(level) != len(values) for level in raw_levels): |
| 123 | raise UnsupportedChartEx("unsupported-chartex-cache") |
| 124 | # ChartEx serializes the innermost level first; the authoring schema |
| 125 | # and workbook writer use the natural outermost-to-innermost order. |
| 126 | payload["levels"] = list(reversed(raw_levels)) |
| 127 | payload["values"] = values |
| 128 | if chart_type == "treemap": |
| 129 | parent_labels = _treemap_parent_labels(series_nodes[0]) |
| 130 | if parent_labels is not None: |
| 131 | payload["parent_label_layout"] = parent_labels |
| 132 | referenced_data_ids = [data_id] |
| 133 | elif chart_type == "histogram": |
| 134 | data_id = _series_data_id(series_nodes[0]) |
| 135 | data = _require_data(data_by_id, data_id) |
| 136 | _reject_dimensions(data, allowed={"numDim"}) |
| 137 | payload["values"] = _numeric_dimension(data, "val") |
| 138 | referenced_data_ids = [data_id] |
| 139 | elif chart_type in {"funnel", "pareto", "waterfall"}: |
| 140 | data_id = _series_data_id(series_nodes[0]) |
| 141 | data = _require_data(data_by_id, data_id) |
| 142 | categories = _string_dimension(data, expected_levels=1)[0] |
| 143 | values = _numeric_dimension(data, "val") |
| 144 | if len(categories) != len(values): |
| 145 | raise UnsupportedChartEx("unsupported-chartex-cache") |
| 146 | payload["categories"] = categories |
| 147 | payload["values"] = values |
| 148 | if chart_type == "waterfall": |
| 149 | payload["subtotals"] = _waterfall_subtotals(series_nodes[0]) |
| 150 | referenced_data_ids = [data_id] |
| 151 | elif chart_type == "box_whisker": |
| 152 | items: list[dict[str, Any]] = [] |
| 153 | referenced_data_ids = [] |
| 154 | for index, series in enumerate(series_nodes, start=1): |
| 155 | data_id = _series_data_id(series) |
| 156 | if data_id in referenced_data_ids: |
| 157 | raise UnsupportedChartEx("unsupported-chartex-data-id") |
| 158 | data = _require_data(data_by_id, data_id) |
| 159 | categories = _string_dimension(data, expected_levels=1)[0] |
| 160 | values = _numeric_dimension(data, "val") |
| 161 | if len(categories) != len(values): |
| 162 | raise UnsupportedChartEx("unsupported-chartex-cache") |
| 163 | items.append({ |
| 164 | "categories": categories, |
| 165 | "name": _series_name(series, index), |
| 166 | "values": values, |
| 167 | }) |
| 168 | referenced_data_ids.append(data_id) |
| 169 | payload["series"] = items |
| 170 | else: # pragma: no cover - _chart_type is closed, keep the invariant explicit. |
| 171 | raise UnsupportedChartEx("unsupported-chartex-type") |
| 172 | |
| 173 | if set(referenced_data_ids) != set(data_by_id): |
| 174 | raise UnsupportedChartEx("unsupported-chartex-data-id") |
| 175 | |
| 176 | legend = chart.find("cx:legend", CX_NS) |
| 177 | if legend is not None: |
| 178 | payload["show_legend"] = True |
| 179 | position = legend.attrib.get("pos") |
| 180 | if position in _LEGEND_POSITIONS: |
| 181 | payload["legend_position"] = position |
| 182 | return payload |
| 183 | |
| 184 | |
| 185 | def _chart_type(series_nodes: list[ET.Element]) -> str: |
| 186 | layouts = [series.attrib.get("layoutId", "") for series in series_nodes] |
| 187 | if len(series_nodes) == 1 and layouts[0] in _HIERARCHY_LAYOUTS | _FLAT_LAYOUTS: |
| 188 | return layouts[0] |
| 189 | if layouts and all(layout == "boxWhisker" for layout in layouts): |
| 190 | return "box_whisker" |
| 191 | if len(series_nodes) == 1 and layouts == ["clusteredColumn"]: |
| 192 | if series_nodes[0].find("cx:layoutPr/cx:binning", CX_NS) is None: |
| 193 | raise UnsupportedChartEx("unsupported-chartex-series") |
| 194 | return "histogram" |
| 195 | if len(series_nodes) == 2 and layouts == ["clusteredColumn", "paretoLine"]: |
| 196 | primary, line = series_nodes |
| 197 | if ( |
| 198 | primary.find("cx:layoutPr/cx:aggregation", CX_NS) is None |
| 199 | or line.attrib.get("ownerIdx") != "0" |
| 200 | or line.find("cx:dataId", CX_NS) is not None |
| 201 | ): |
| 202 | raise UnsupportedChartEx("unsupported-chartex-series") |
| 203 | return "pareto" |
| 204 | raise UnsupportedChartEx("unsupported-chartex-type") |
| 205 | |
| 206 | |
| 207 | def _data_parts(chart_data: ET.Element) -> dict[int, ET.Element]: |
| 208 | result: dict[int, ET.Element] = {} |
| 209 | for data in _children(chart_data, "data"): |
| 210 | data_id = _nonnegative_int(data.attrib.get("id"), "unsupported-chartex-data-id") |
| 211 | if data_id in result: |
| 212 | raise UnsupportedChartEx("unsupported-chartex-data-id") |
| 213 | result[data_id] = data |
| 214 | if not result: |
| 215 | raise UnsupportedChartEx("unsupported-chartex-data-id") |
| 216 | return result |
| 217 | |
| 218 | |
| 219 | def _series_data_id(series: ET.Element) -> int: |
| 220 | data_ids = _children(series, "dataId") |
| 221 | if len(data_ids) != 1: |
| 222 | raise UnsupportedChartEx("unsupported-chartex-data-id") |
| 223 | return _nonnegative_int( |
| 224 | data_ids[0].attrib.get("val"), |
| 225 | "unsupported-chartex-data-id", |
| 226 | ) |
| 227 | |
| 228 | |
| 229 | def _require_data(data_by_id: dict[int, ET.Element], data_id: int) -> ET.Element: |
| 230 | data = data_by_id.get(data_id) |
| 231 | if data is None: |
| 232 | raise UnsupportedChartEx("unsupported-chartex-data-id") |
| 233 | return data |
| 234 | |
| 235 | |
| 236 | def _string_dimension( |
| 237 | data: ET.Element, |
| 238 | *, |
| 239 | expected_levels: int | None, |
| 240 | ) -> list[list[str]]: |
| 241 | _reject_dimensions(data, allowed={"strDim", "numDim"}) |
| 242 | dimensions = [ |
| 243 | child |
| 244 | for child in _children(data, "strDim") |
| 245 | if child.attrib.get("type") == "cat" |
| 246 | ] |
| 247 | if len(dimensions) != 1 or len(_children(data, "strDim")) != 1: |
| 248 | raise UnsupportedChartEx("unsupported-chartex-dimension") |
| 249 | levels = _children(dimensions[0], "lvl") |
| 250 | if not levels or (expected_levels is not None and len(levels) != expected_levels): |
| 251 | raise UnsupportedChartEx("unsupported-chartex-dimension") |
| 252 | return [_level_values(level, numeric=False) for level in levels] |
| 253 | |
| 254 | |
| 255 | def _numeric_dimension(data: ET.Element, dim_type: str) -> list[int | float]: |
| 256 | dimensions = [ |
| 257 | child |
| 258 | for child in _children(data, "numDim") |
| 259 | if child.attrib.get("type") == dim_type |
| 260 | ] |
| 261 | if len(dimensions) != 1 or len(_children(data, "numDim")) != 1: |
| 262 | raise UnsupportedChartEx("unsupported-chartex-dimension") |
| 263 | levels = _children(dimensions[0], "lvl") |
| 264 | if len(levels) != 1: |
| 265 | raise UnsupportedChartEx("unsupported-chartex-dimension") |
| 266 | return _level_values(levels[0], numeric=True) |
| 267 | |
| 268 | |
| 269 | def _reject_dimensions(data: ET.Element, *, allowed: set[str]) -> None: |
| 270 | for child in data: |
| 271 | name = _local_name(child.tag) |
| 272 | if name.endswith("Dim") and name not in allowed: |
| 273 | raise UnsupportedChartEx("unsupported-chartex-dimension") |
| 274 | |
| 275 | |
| 276 | def _level_values( |
| 277 | level: ET.Element, |
| 278 | *, |
| 279 | numeric: bool, |
| 280 | ) -> list[Any]: |
| 281 | point_count = _nonnegative_int( |
| 282 | level.attrib.get("ptCount"), |
| 283 | "unsupported-chartex-cache", |
| 284 | ) |
| 285 | points: dict[int, Any] = {} |
| 286 | for point in _children(level, "pt"): |
| 287 | point_index = _nonnegative_int( |
| 288 | point.attrib.get("idx"), |
| 289 | "unsupported-chartex-cache", |
| 290 | ) |
| 291 | if point_index >= point_count or point_index in points or list(point): |
| 292 | raise UnsupportedChartEx("unsupported-chartex-cache") |
| 293 | raw_value = point.text or "" |
| 294 | points[point_index] = ( |
| 295 | _numeric_value(raw_value) |
| 296 | if numeric |
| 297 | else raw_value |
| 298 | ) |
| 299 | if ( |
| 300 | len(points) != point_count |
| 301 | or any(index not in points for index in range(point_count)) |
| 302 | ): |
| 303 | raise UnsupportedChartEx("unsupported-chartex-cache") |
| 304 | return [points[index] for index in range(point_count)] |
| 305 | |
| 306 | |
| 307 | def _numeric_value(raw_value: str) -> int | float: |
| 308 | if not raw_value.strip(): |
| 309 | raise UnsupportedChartEx("unsupported-chartex-cache") |
| 310 | try: |
| 311 | number = float(raw_value) |
| 312 | except (OverflowError, ValueError): |
| 313 | raise UnsupportedChartEx("unsupported-chartex-cache") from None |
| 314 | if not math.isfinite(number): |
| 315 | raise UnsupportedChartEx("unsupported-chartex-cache") |
| 316 | return int(number) if number.is_integer() else number |
| 317 | |
| 318 | |
| 319 | def _waterfall_subtotals(series: ET.Element) -> list[int]: |
| 320 | subtotals: list[int] = [] |
| 321 | for item in series.findall("cx:layoutPr/cx:subtotals/cx:idx", CX_NS): |
| 322 | subtotals.append( |
| 323 | _nonnegative_int(item.attrib.get("val"), "unsupported-chartex-cache") |
| 324 | ) |
| 325 | return subtotals |
| 326 | |
| 327 | |
| 328 | def _treemap_parent_labels(series: ET.Element) -> str | None: |
| 329 | parent = series.find("cx:layoutPr/cx:parentLabelLayout", CX_NS) |
| 330 | if parent is None: |
| 331 | return None |
| 332 | value = parent.attrib.get("val") |
| 333 | return value if value in {"banner", "none", "overlapping"} else None |
| 334 | |
| 335 | |
| 336 | def _series_name(series: ET.Element, index: int) -> str: |
| 337 | value = series.findtext("cx:tx/cx:txData/cx:v", default="", namespaces=CX_NS) |
| 338 | return value or f"Series {index}" |
| 339 | |
| 340 | |
| 341 | def _resolved_chart_colors( |
| 342 | chart_part: PartRef, |
| 343 | pkg: OoxmlPackage, |
| 344 | palette: ColorPalette | None, |
| 345 | ) -> list[str]: |
| 346 | """Resolve the base color cycle; any style failure normalizes silently.""" |
| 347 | try: |
| 348 | targets = [ |
| 349 | info.get("target") |
| 350 | for info in chart_part.rels.values() |
| 351 | if info.get("type") == CHART_COLOR_STYLE_REL_TYPE |
| 352 | and not info.get("external") |
| 353 | and info.get("target") |
| 354 | ] |
| 355 | if len(targets) != 1: |
| 356 | return [] |
| 357 | colors_part = pkg.load_part(str(targets[0])) |
| 358 | if colors_part is None: |
| 359 | return [] |
| 360 | color_nodes = [ |
| 361 | child |
| 362 | for child in colors_part.xml |
| 363 | if _namespace(child.tag) == NS["a"] |
| 364 | and _local_name(child.tag) in COLOR_TAGS |
| 365 | ] |
| 366 | if not color_nodes: |
| 367 | return [] |
| 368 | colors: list[str] = [] |
| 369 | for color_node in color_nodes: |
| 370 | color, _alpha = resolve_color(color_node, palette) |
| 371 | if color is None: |
| 372 | return [] |
| 373 | colors.append(color) |
| 374 | return colors |
| 375 | except (AttributeError, OverflowError, RuntimeError, TypeError, ValueError): |
| 376 | return [] |
| 377 | |
| 378 | |
| 379 | def _one_child(parent: ET.Element, name: str, status: str) -> ET.Element: |
| 380 | children = _children(parent, name) |
| 381 | if len(children) != 1: |
| 382 | raise UnsupportedChartEx(status) |
| 383 | return children[0] |
| 384 | |
| 385 | |
| 386 | def _children(parent: ET.Element, name: str) -> list[ET.Element]: |
| 387 | return [ |
| 388 | child |
| 389 | for child in parent |
| 390 | if child.tag == f"{{{CHARTEX_URI}}}{name}" |
| 391 | ] |
| 392 | |
| 393 | |
| 394 | def _nonnegative_int(raw_value: str | None, status: str) -> int: |
| 395 | if raw_value is None or _INT_TOKEN_RE.fullmatch(raw_value) is None: |
| 396 | raise UnsupportedChartEx(status) |
| 397 | value = int(raw_value) |
| 398 | if value < 0: |
| 399 | raise UnsupportedChartEx(status) |
| 400 | return value |
| 401 | |
| 402 | |
| 403 | def _bounds_payload(xfrm: Xfrm) -> dict[str, int | float]: |
| 404 | return { |
| 405 | "height": _round_payload_number(xfrm.h), |
| 406 | "width": _round_payload_number(xfrm.w), |
| 407 | "x": _round_payload_number(xfrm.x), |
| 408 | "y": _round_payload_number(xfrm.y), |
| 409 | } |
| 410 | |
| 411 | |
| 412 | def _round_payload_number(value: float) -> int | float: |
| 413 | rounded = round(float(value), 3) |
| 414 | return int(rounded) if rounded.is_integer() else rounded |
| 415 | |
| 416 | |
| 417 | def _namespace(tag: str) -> str: |
| 418 | return tag[1:].split("}", 1)[0] if tag.startswith("{") else "" |
| 419 | |
| 420 | |
| 421 | def _local_name(tag: str) -> str: |
| 422 | return tag.rsplit("}", 1)[-1] if "}" in tag else tag |
| 423 |