162 lines
6.0 KiB
Python
162 lines
6.0 KiB
Python
"""Lazy headless rendering for immutable schedule reports."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import io
|
|
import math
|
|
import struct
|
|
from collections.abc import Callable
|
|
from typing import Any, Final, cast
|
|
|
|
from comfyui_sigmax.core import (
|
|
ScheduleComparisonReport,
|
|
ScheduleContractError,
|
|
ScheduleReport,
|
|
)
|
|
|
|
_FORMATS: Final = frozenset({"png", "svg"})
|
|
|
|
|
|
class SchedulePlotDependencyError(RuntimeError):
|
|
"""Raised when the explicitly optional plotting extra is unavailable."""
|
|
|
|
|
|
def _format(value: object) -> str:
|
|
if not isinstance(value, str) or value not in _FORMATS:
|
|
raise ScheduleContractError("plot format must be png or svg")
|
|
return value
|
|
|
|
|
|
def _decode_token(value: object) -> float:
|
|
if not isinstance(value, dict) or set(value) != {"bits", "precision"}:
|
|
raise ScheduleContractError("plot input contains an invalid typed float")
|
|
bits = value["bits"]
|
|
precision = value["precision"]
|
|
if precision == "float32":
|
|
expected_length = 8
|
|
format_code = ">f"
|
|
elif precision == "float64":
|
|
expected_length = 16
|
|
format_code = ">d"
|
|
else:
|
|
raise ScheduleContractError("plot input precision is unsupported")
|
|
if (
|
|
not isinstance(bits, str)
|
|
or len(bits) != expected_length
|
|
or any(character not in "0123456789abcdef" for character in bits)
|
|
):
|
|
raise ScheduleContractError("plot input float bits are invalid")
|
|
number = struct.unpack(format_code, bytes.fromhex(bits))[0]
|
|
if not math.isfinite(number):
|
|
raise ScheduleContractError("plot input must be finite")
|
|
return 0.0 if number == 0.0 else number
|
|
|
|
|
|
def _matplotlib() -> tuple[Callable[..., Any], Callable[..., Any]]:
|
|
try:
|
|
figure_module = importlib.import_module("matplotlib.figure")
|
|
agg_module = importlib.import_module("matplotlib.backends.backend_agg")
|
|
except (ImportError, ModuleNotFoundError) as exc:
|
|
raise SchedulePlotDependencyError(
|
|
"schedule plotting requires the optional extra: pip install 'comfyui-sigmax[plot]'"
|
|
) from exc
|
|
figure = getattr(figure_module, "Figure", None)
|
|
canvas = getattr(agg_module, "FigureCanvasAgg", None)
|
|
if not callable(figure) or not callable(canvas):
|
|
raise SchedulePlotDependencyError("installed Matplotlib lacks the required headless API")
|
|
return cast(Callable[..., Any], figure), cast(Callable[..., Any], canvas)
|
|
|
|
|
|
def _save(figure: Any, *, image_format: str) -> bytes:
|
|
payload = io.BytesIO()
|
|
metadata: dict[str, object] = (
|
|
{"Date": None, "Creator": "ComfyUI-Sigmax"}
|
|
if image_format == "svg"
|
|
else {"Software": "ComfyUI-Sigmax"}
|
|
)
|
|
try:
|
|
figure.savefig(
|
|
payload,
|
|
format=image_format,
|
|
dpi=120,
|
|
metadata=metadata,
|
|
)
|
|
rendered = payload.getvalue()
|
|
finally:
|
|
figure.clear()
|
|
payload.close()
|
|
signature = b"\x89PNG\r\n\x1a\n" if image_format == "png" else b"<?xml"
|
|
if not rendered.startswith(signature):
|
|
raise ScheduleContractError("plot renderer returned an invalid image payload")
|
|
return rendered
|
|
|
|
|
|
def render_schedule_plot(
|
|
report: ScheduleReport,
|
|
*,
|
|
image_format: str = "png",
|
|
) -> bytes:
|
|
"""Render one report to in-memory PNG/SVG presentation bytes."""
|
|
|
|
selected_format = _format(image_format)
|
|
if not isinstance(report, ScheduleReport):
|
|
raise ScheduleContractError("schedule plot requires a ScheduleReport")
|
|
projection = report.projection()
|
|
samples = cast(list[dict[str, object]], projection["samples"])
|
|
indices = [cast(int, sample["index"]) for sample in samples]
|
|
sigmas = [_decode_token(sample["sigma"]) for sample in samples]
|
|
figure_type, canvas_type = _matplotlib()
|
|
figure = figure_type(figsize=(8.0, 5.0), dpi=120)
|
|
canvas_type(figure)
|
|
axis = figure.subplots()
|
|
axis.plot(indices, sigmas, marker="o", linewidth=1.8, label="sigma")
|
|
axis.set_title("Sigma schedule")
|
|
axis.set_xlabel("Terminal-inclusive sigma index")
|
|
axis.set_ylabel(f"Sigma ({projection['domain']})")
|
|
axis.grid(True, alpha=0.25)
|
|
axis.legend()
|
|
return _save(figure, image_format=selected_format)
|
|
|
|
|
|
def render_schedule_comparison_plot(
|
|
report: ScheduleComparisonReport,
|
|
*,
|
|
image_format: str = "png",
|
|
) -> bytes:
|
|
"""Render two aligned schedules and their absolute differences."""
|
|
|
|
selected_format = _format(image_format)
|
|
if not isinstance(report, ScheduleComparisonReport):
|
|
raise ScheduleContractError("comparison plot requires a ScheduleComparisonReport")
|
|
projection = report.projection()
|
|
if projection["comparable"] is not True:
|
|
raise ScheduleContractError("comparison plot requires comparable schedules")
|
|
samples = cast(list[dict[str, object]], projection["samples"])
|
|
indices = [cast(int, sample["index"]) for sample in samples]
|
|
sigmas_a = [_decode_token(sample["sigma_a"]) for sample in samples]
|
|
sigmas_b = [_decode_token(sample["sigma_b"]) for sample in samples]
|
|
absolute = [_decode_token(sample["absolute_difference"]) for sample in samples]
|
|
figure_type, canvas_type = _matplotlib()
|
|
figure = figure_type(figsize=(8.0, 7.0), dpi=120)
|
|
canvas_type(figure)
|
|
schedule_axis, difference_axis = figure.subplots(2, 1, sharex=True)
|
|
schedule_axis.plot(indices, sigmas_a, marker="o", linewidth=1.6, label="schedule A")
|
|
schedule_axis.plot(indices, sigmas_b, marker="s", linewidth=1.6, label="schedule B")
|
|
schedule_axis.set_title("Sigma schedule comparison")
|
|
schedule_axis.set_ylabel("Sigma")
|
|
schedule_axis.grid(True, alpha=0.25)
|
|
schedule_axis.legend()
|
|
difference_axis.plot(indices, absolute, color="#b34747", linewidth=1.6)
|
|
difference_axis.set_xlabel("Terminal-inclusive sigma index")
|
|
difference_axis.set_ylabel("Absolute difference")
|
|
difference_axis.grid(True, alpha=0.25)
|
|
return _save(figure, image_format=selected_format)
|
|
|
|
|
|
__all__ = [
|
|
"SchedulePlotDependencyError",
|
|
"render_schedule_comparison_plot",
|
|
"render_schedule_plot",
|
|
]
|