printing

Column dataclass

Specification for a single column in the iteration table.

Source code in openscvx/utils/printing.py

@dataclass
class Column:
    """Specification for a single column in the iteration table."""

    key: str  # Key in emission data dict
    header: str  # Column header text
    width: int  # Column width
    fmt: str  # Format string for values
    color_fn: Optional[Callable[[Any, Any, dict], Optional[str]]] = None
    min_verbosity: int = field(default=Verbosity.MINIMAL)  # Minimum verbosity to show

Verbosity

Bases: IntEnum

Verbosity levels for iteration table output.

Source code in openscvx/utils/printing.py

class Verbosity(IntEnum):
    """Verbosity levels for iteration table output."""

    MINIMAL = 1  # Core metrics only (iter, cost, status)
    STANDARD = 2  # + timing, penalty terms
    FULL = 3  # + autotuning diagnostics (J_nonlin, reductions, etc.)

build_header_format(columns: list[Column]) -> str

Generate header format string from active columns.

Source code in openscvx/utils/printing.py

def build_header_format(columns: list[Column]) -> str:
    """Generate header format string from active columns."""
    return " │ ".join(f"{{:^{col.width}}}" for col in columns)

build_separator(columns: list[Column]) -> str

Generate separator line matching the total width of active columns.

Source code in openscvx/utils/printing.py

def build_separator(columns: list[Column]) -> str:
    """Generate separator line matching the total width of active columns."""
    total_width = sum(col.width for col in columns) + 3 * (len(columns) - 1)
    return "─" * total_width

color_J_nonlin(value: Any, params: Any, data: dict) -> Optional[str]

Color J_nonlin green if positive, red otherwise.

Source code in openscvx/utils/printing.py

def color_J_nonlin(value: Any, params: Any, data: dict) -> Optional[str]:
    """Color J_nonlin green if positive, red otherwise."""
    return "green" if value > 0 else "red"

color_J_tr(value: Any, params: Any, data: dict) -> Optional[str]

Color J_tr green if within tolerance, red otherwise.

Source code in openscvx/utils/printing.py

def color_J_tr(value: Any, params: Any, data: dict) -> Optional[str]:
    """Color J_tr green if within tolerance, red otherwise."""
    ep = data.get("ep_tr")
    if ep is None:
        return None
    return col_pos if value <= ep else col_neg

color_J_vb(value: Any, params: Any, data: dict) -> Optional[str]

Color J_vb green if within tolerance, red otherwise.

Source code in openscvx/utils/printing.py

def color_J_vb(value: Any, params: Any, data: dict) -> Optional[str]:
    """Color J_vb green if within tolerance, red otherwise."""
    ep = data.get("ep_vb")
    if ep is None:
        return None
    return col_pos if value <= ep else col_neg

color_J_vc(value: Any, params: Any, data: dict) -> Optional[str]

Color J_vc green if within tolerance, red otherwise.

Source code in openscvx/utils/printing.py

def color_J_vc(value: Any, params: Any, data: dict) -> Optional[str]:
    """Color J_vc green if within tolerance, red otherwise."""
    ep = data.get("ep_vc")
    if ep is None:
        return None
    return col_pos if value <= ep else col_neg

color_acceptance_ratio(value: Any, params: Any, data: dict) -> Optional[str]

Color acceptance ratio based on success level.

<= 0.1: red (unsuccessful) 0.1 < ratio <= 0.8: somewhat successful (green) 0.8 < ratio <= 1.5: very successful (blue)

1.5: overly successful (magenta)

Source code in openscvx/utils/printing.py

def color_acceptance_ratio(value: Any, params: Any, data: dict) -> Optional[str]:
    """Color acceptance ratio based on success level.

    <= 0.1: red (unsuccessful)
    0.1 < ratio <= 0.8: somewhat successful (green)
    0.8 < ratio <= 1.5: very successful (blue)
    > 1.5: overly successful (magenta)
    """
    if value <= 0.1:
        return "red"
    elif value <= 0.8:
        return "green"
    elif value <= 1.5:
        return "blue"
    else:
        return "magenta"

color_adaptive_state(value: Any, params: Any, data: dict) -> Optional[str]

Color adaptive state green if acceptable, red otherwise.

Source code in openscvx/utils/printing.py

def color_adaptive_state(value: Any, params: Any, data: dict) -> Optional[str]:
    """Color adaptive state green if acceptable, red otherwise."""
    acceptable_states = ["Accept Constant", "Accept Higher", "Accept Lower", "Initial"]
    return col_pos if value in acceptable_states else col_neg

color_prob_stat(value: Any, params: Any, data: dict) -> Optional[str]

Color solver status green if optimal, red otherwise.

Source code in openscvx/utils/printing.py

def color_prob_stat(value: Any, params: Any, data: dict) -> Optional[str]:
    """Color solver status green if optimal, red otherwise."""
    return col_pos if value == "optimal" else col_neg

footer(columns: list[Column]) -> None

Print the table footer.

Parameters:

Name	Type	Description	Default
columns	list[Column]	List of Column specs defining the table structure	required

Source code in openscvx/utils/printing.py

def footer(columns: list[Column]) -> None:
    """Print the table footer.

    Args:
        columns: List of Column specs defining the table structure
    """
    separator = build_separator(columns)
    print(colored(separator))

format_value(col: Column, value: Any, params: Any, data: dict) -> str

Format a single value and apply coloring if needed.

Source code in openscvx/utils/printing.py

def format_value(col: Column, value: Any, params: Any, data: dict) -> str:
    """Format a single value and apply coloring if needed."""
    # Handle None values
    if value is None:
        value = 0.0

    # Format the value
    formatted = col.fmt.format(value)

    # Pad to column width BEFORE applying color (ANSI codes break alignment)
    formatted = f"{formatted:^{col.width}}"

    # Apply coloring if a color function is defined
    if col.color_fn is not None:
        color = col.color_fn(value, params, data)
        if color is not None:
            formatted = colored(formatted, color)

    return formatted

header(columns: list[Column]) -> None

Print the table header for the given columns.

Source code in openscvx/utils/printing.py

def header(columns: list[Column]) -> None:
    """Print the table header for the given columns."""
    separator = build_separator(columns)
    header_fmt = build_header_format(columns)

    print(colored(separator))
    print(header_fmt.format(*[col.header for col in columns]))
    print(colored(separator))

intermediate(print_queue: queue.Queue, params: Any, columns: list[Column]) -> None

Process and print iteration data from the queue.

This function runs in a loop, reading data from the print queue and displaying formatted iteration rows.

Parameters:

Name	Type	Description	Default
print_queue	Queue	Queue containing iteration data dicts	required
params	Any	Settings object (used for color threshold comparisons)	required
columns	list[Column]	List of Column specs defining the table structure	required

Source code in openscvx/utils/printing.py

def intermediate(print_queue: queue.Queue, params: Any, columns: list[Column]) -> None:
    """Process and print iteration data from the queue.

    This function runs in a loop, reading data from the print queue and
    displaying formatted iteration rows.

    Args:
        print_queue: Queue containing iteration data dicts
        params: Settings object (used for color threshold comparisons)
        columns: List of Column specs defining the table structure
    """
    hz = 30.0
    separator = build_separator(columns)

    while True:
        t_start = time.time()
        try:
            data = print_queue.get(timeout=1.0 / hz)

            # Truncate prob_stat if it's a longer string (e.g., "infeasible" -> "i")
            prob_stat = data.get("prob_stat", "")
            if prob_stat.startswith("inf"):
                data["prob_stat"] = "i"

            # Remove bottom separator and header (2 lines)
            if data["iter"] != 1:
                sys.stdout.write("\x1b[1A\x1b[2K\x1b[1A\x1b[2K")

            # Print the data row
            print_row(columns, data, params)

            # Print separator and header (footer() will add closing separator)
            print(colored(separator))
            header_fmt = build_header_format(columns)
            print(header_fmt.format(*[col.header for col in columns]))

        except queue.Empty:
            pass
        time.sleep(max(0.0, 1.0 / hz - (time.time() - t_start)))

print_batch_results_summary(result: OptimizationResults, timing_post: float, timing_init: Optional[float], timing_solve: Optional[float], *, timing_compile: Optional[float] = None) -> None

Print a results summary box for a batched post-processed solve.

Mirrors :func:print_results_summary for the output of :meth:~openscvx.problem.Problem.post_process_batched.

Parameters:

Name	Type	Description	Default
result	OptimizationResults	Post-processed :class:~openscvx.algorithms.OptimizationResults with x_full, t_full, cost, and ctcs_violation set.	required
timing_post	float	Propagation wall-clock time in seconds.	required
timing_init	Optional[float]	Initialize wall-clock time in seconds (None if unknown).	required
timing_solve	Optional[float]	Batched-solve execution time in seconds (None if unknown). Compile time is excluded; see timing_compile.	required
timing_compile	Optional[float]	XLA/AOT compile time in seconds for the batched solve (None when the compiled artifact was loaded from cache).	None

Source code in openscvx/utils/printing.py

def print_batch_results_summary(
    result: "OptimizationResults",
    timing_post: float,
    timing_init: Optional[float],
    timing_solve: Optional[float],
    *,
    timing_compile: Optional[float] = None,
) -> None:
    """Print a results summary box for a batched post-processed solve.

    Mirrors :func:`print_results_summary` for the output of
    :meth:`~openscvx.problem.Problem.post_process_batched`.

    Args:
        result: Post-processed :class:`~openscvx.algorithms.OptimizationResults`
            with ``x_full``, ``t_full``, ``cost``, and ``ctcs_violation`` set.
        timing_post:    Propagation wall-clock time in seconds.
        timing_init:    Initialize wall-clock time in seconds (``None`` if unknown).
        timing_solve:   Batched-solve execution time in seconds (``None`` if unknown).
            Compile time is excluded; see ``timing_compile``.
        timing_compile: XLA/AOT compile time in seconds for the batched solve
            (``None`` when the compiled artifact was loaded from cache).
    """
    converged = np.asarray(result.converged, dtype=bool).reshape(-1)
    B = converged.shape[0]
    n_ok = int(converged.sum())
    n_bad = B - n_ok

    t_full_arr = np.asarray(result.t_full)  # (B, T)
    T = t_full_arr.shape[1]
    dt = float(t_full_arr[0, 1] - t_full_arr[0, 0]) if T > 1 else 0.0

    t_finals = t_full_arr[:, -1]
    if converged.any():
        t_f_conv = t_finals[converged]
        t_f_str = (
            f"{t_f_conv.mean():.3f} s  (min {t_f_conv.min():.3f} s, max {t_f_conv.max():.3f} s)"
        )
    else:
        t_f_str = "N/A (no converged solutions)"

    cost_arr = np.asarray(result.cost)
    if converged.any():
        c_conv = cost_arr[converged]
        cost_str = f"{c_conv.mean():.4f}  (min {c_conv.min():.4f}, max {c_conv.max():.4f})"
    else:
        cost_str = "N/A"

    ctcs = result.ctcs_violation
    if ctcs is not None and converged.any():
        ctcs_arr = np.abs(np.asarray(ctcs)[converged])
        ctcs_str = f"{ctcs_arr.max():.2e}"
    else:
        ctcs_str = "N/A"

    t_init = timing_init or 0.0
    t_compile = timing_compile or 0.0
    t_solve = timing_solve or 0.0
    total_time = t_init + t_compile + t_solve + timing_post

    lines = [
        "Batch Results Summary",
        f"Batch size:              {B}",
        f"Converged:               {n_ok} / {B}  ({100 * n_ok / B:.0f}%)",
        f"Diverged:                {n_bad} / {B}",
        "SEP",
        f"Initialize time:         {t_init:.3f} s",
        f"Compile time:            {t_compile:.3f} s",
        f"Batched solve time:      {t_solve:.3f} s",
        f"Propagation time:        {timing_post:.3f} s",
        f"Total time:              {total_time:.3f} s",
        "SEP",
        f"Propagated steps:        {T}  (dt \u2248 {dt:.4f} s)",
        f"Terminal time (conv.):   {t_f_str}",
        f"Cost (conv.):            {cost_str}",
        f"Max CTCS violation:      {ctcs_str}",
    ]
    print_summary_box(lines, "Batch Results Summary")

print_problem_summary(settings: Config, lowered: Any, solver: ConvexSolver, algorithm: Algorithm, discretizer: Discretizer) -> None

Print the problem summary box.

Parameters:

Name	Type	Description	Default
settings	Config	Configuration settings containing problem information	required
lowered	Any	LoweredProblem from lower_symbolic_problem()	required
solver	ConvexSolver	Initialized ConvexSolver with built problem	required
algorithm	Algorithm	Algorithm instance (e.g., PenalizedTrustRegion)	required
discretizer	Discretizer	Discretizer instance (e.g., LinearizeDiscretize)	required

Source code in openscvx/utils/printing.py

def print_problem_summary(
    settings: "Config",
    lowered: Any,
    solver: "ConvexSolver",
    algorithm: "Algorithm",
    discretizer: "Discretizer",
) -> None:
    """
    Print the problem summary box.

    Args:
        settings: Configuration settings containing problem information
        lowered: LoweredProblem from lower_symbolic_problem()
        solver: Initialized ConvexSolver with built problem
        algorithm: Algorithm instance (e.g., PenalizedTrustRegion)
        discretizer: Discretizer instance (e.g., LinearizeDiscretize)
    """
    n_nodal_convex = len(lowered.cvxpy_constraints.constraints)
    n_nodal_nonconvex = len(lowered.jax_constraints.nodal)
    n_ctcs = len(lowered.jax_constraints.ctcs)
    n_augmented = settings.sim.n_states - settings.sim.true_state_slice.stop

    # Get solver statistics
    stats = solver.get_stats()
    n_cvx_variables = stats["n_variables"]
    n_cvx_parameters = stats["n_parameters"]
    n_cvx_constraints = stats["n_constraints"]

    # Get JAX backend information
    jax_backend = jax.devices()[0].platform.upper()
    jax_version = jax.__version__
    float_dtype = "float64" if jax.config.read("jax_enable_x64") else "float32"

    algorithm_name = type(algorithm).__name__
    autotuner = getattr(algorithm, "autotuner", None)
    autotuner_name = type(autotuner).__name__ if autotuner is not None else "N/A"
    discretization_name = type(discretizer).__name__

    solver_name = type(solver).__name__
    cvx_solver_label = getattr(solver, "cvx_solver", None)
    solver_line = (
        f"Solver: {solver_name} ({cvx_solver_label})"
        if cvx_solver_label is not None
        else f"Solver: {solver_name}"
    )

    lines = [
        "Problem Summary",
        (
            f"Dimensions: {settings.sim.n_states} states ({n_augmented} aug),"
            f" {settings.sim.n_controls} controls, {settings.sim.n} nodes"
        ),
        f"Constraints: {n_nodal_convex} conv, {n_nodal_nonconvex} nonconv, {n_ctcs} ctcs",
        (
            f"Subproblem: {n_cvx_variables} vars, {n_cvx_parameters} params,"
            f" {n_cvx_constraints} constraints"
        ),
        "SEP",
        f"Algorithm: {algorithm_name}",
        f"Autotuner: {autotuner_name}",
        f"Discretizer: {discretization_name} ({discretizer.ode_solver})",
        "SEP",
        solver_line,
        f"Float Dtype: {float_dtype}",
        f"JAX Backend: {jax_backend} (v{jax_version})",
    ]

    print_summary_box(lines, "Problem Summary")

print_results_summary(result: OptimizationResults, timing_post, timing_init, timing_solve)

Print the results summary box.

Parameters:

Name	Type	Description	Default
result	OptimizationResults	Optimization results object	required
timing_post	float	Post-processing time	required
timing_init	float	Initialization time	required
timing_solve	float	Solve time	required

Source code in openscvx/utils/printing.py

def print_results_summary(result: "OptimizationResults", timing_post, timing_init, timing_solve):
    """
    Print the results summary box.

    Args:
        result (OptimizationResults): Optimization results object
        timing_post (float): Post-processing time
        timing_init (float): Initialization time
        timing_solve (float): Solve time
    """
    cost = result.get("cost", 0.0)
    ctcs_violation = result.get("ctcs_violation", 0.0)

    # Convert numpy arrays to scalars for formatting
    if hasattr(cost, "item"):
        cost = cost.item()

    # Handle CTCS violation - display as 1D array
    if hasattr(ctcs_violation, "size"):
        if ctcs_violation.size == 1:
            ctcs_violation_str = f"[{ctcs_violation.item():.2e}]"
        else:
            # Display as 1D array
            ctcs_violation_str = f"[{', '.join([f'{v:.2e}' for v in ctcs_violation])}]"
    else:
        ctcs_violation_str = f"[{ctcs_violation:.2e}]"

    # Calculate total computation time
    total_time = (timing_init or 0.0) + (timing_solve or 0.0) + timing_post

    lines = [
        "Results Summary",
        f"Cost: {cost:.6f}",
        f"CTCS Constraint Violation: {ctcs_violation_str}",
        f"Preprocessing Time: {timing_init or 0.0:.3f}s",
        f"Main Solve Time: {timing_solve or 0.0:.3f}s",
        f"Post-processing Time: {timing_post:.3f}s",
        f"Total Computation Time: {total_time:.3f}s",
    ]

    print_summary_box(lines, "Results Summary")

print_row(columns: list[Column], data: dict, params: Any) -> None

Print a single data row for the given columns.

Source code in openscvx/utils/printing.py

def print_row(columns: list[Column], data: dict, params: Any) -> None:
    """Print a single data row for the given columns."""
    values = [format_value(col, data.get(col.key), params, data) for col in columns]
    print(" │ ".join(values))

print_summary_box(lines, title='Summary')

Print a centered summary box with the given lines.

Parameters:

Name	Type	Description	Default
lines	list	List of strings to display in the box	required
title	str	Title for the box (default: "Summary")	'Summary'

Source code in openscvx/utils/printing.py

def print_summary_box(lines, title="Summary"):
    """
    Print a centered summary box with the given lines.

    Args:
        lines (list): List of strings to display in the box
        title (str): Title for the box (default: "Summary")
    """
    # Find the longest line (excluding the title which will be handled separately)
    content_lines = lines[1:] if len(lines) > 1 else []
    max_content_width = max(len(line) for line in content_lines) if content_lines else 0
    title_width = len(title)

    # Box width should accommodate both title and content
    box_width = max(max_content_width, title_width) + 4  # Add padding for the box borders

    # Center with respect to the 89-character horizontal lines in io.py
    total_width = 89
    if box_width <= total_width:
        indent = (total_width - box_width) // 2
    else:
        # If box is wider than 89 chars, use a smaller fixed indentation
        indent = 2

    # Print the box with dynamic width and centering
    print(f"\n{' ' * indent}╭{'─' * box_width}╮")
    print(f"{' ' * indent}│ {title:^{box_width - 2}} │")
    print(f"{' ' * indent}├{'─' * box_width}┤")
    for line in content_lines:
        if line == "SEP":
            print(f"{' ' * indent}├{'─' * box_width}┤")
            continue
        print(f"{' ' * indent}│ {line:<{box_width - 2}} │")
    print(f"{' ' * indent}╰{'─' * box_width}╯\n")