ddp_utils.console.console¶

import ddp_utils.console.console

Coordinate terminal writes, intercepted streams, and live footer layouts.

The console owns the terminal lock, stdout and stderr interception, layout and standalone-renderer registries, and addressed footer repainting. Logical ANSI strings remain the public compatibility boundary; internally they are converted to Rich segments for cell-aware wrapping. The live area is anchored to an absolute terminal row, avoiding footer-height guesses and repeated output.

The implementation deliberately does not use rich.live.Live; terminal ownership stays with Console.

Examples

Coordinate ordinary output with a temporary live layout:

from ddp_utils.console.console import Console

with Console(auto_install_proxy=True) as console:
    layout = console.create_layout(title="Import")
    layout.append_log("started")

class ddp_utils.console.console.ConsoleStreamProxy(console: Console, stream_name: str)¶

Bases: TextIOBase

Route text-stream writes through an owning Console.

The proxy preserves the standard stream surface needed by print and sys.stdout.write while allowing the console to serialize user output with footer repainting.

Examples

Create a stdout proxy for an existing console:

proxy = ConsoleStreamProxy(console, "stdout")

Bind the proxy to a console and one named real stream.

Parameters

Name	Type	Description
console	Console	Console coordinating proxy writes.
stream_name	str	Stream selector, normally `"stdout"` or `"stderr"`.

Examples

Proxy stderr through the console coordinator:

proxy = ConsoleStreamProxy(console, "stderr")

property encoding: str¶

Return the underlying stream encoding with a UTF-8 fallback.

Returns

Non-empty encoding name.

Examples

Inspect the encoding expected by intercepted writes:

encoding = proxy.encoding

isatty() → bool¶

Return whether the underlying real stream is a TTY.

Returns

Type	Description
bool	`True` when the selected real stream reports terminal support.

Examples

Preserve terminal capability checks performed by clients:

interactive = proxy.isatty()

fileno() → int¶

Return the underlying real stream’s file descriptor, or raise OSError if it has none.

Returns

Type	Description
int	Operating-system file descriptor from the selected real stream.

Raises

Exception	Description
OSError	The underlying stream does not expose a file descriptor.

Examples

Pass the intercepted stream descriptor to terminal code:

descriptor = proxy.fileno()

write(s: str) → int¶

Write through the console or directly while bypass mode is active.

Parameters

Name	Type	Description
s	str	Text-like value converted to `str` before routing.

Returns

Type	Description
int	Number of characters in the routed string.

Examples

Route a normal stream write through the console:

written = proxy.write("processing\n")

class ddp_utils.console.console.ConsoleLayout(console: Console, title: str | None = None, max_logs: int | None = None, on_finish: str = 'keep', parent: ConsoleLayout | None = None, indent_step: int = 2)¶

Bases: object

Represent one independently rendered block owned by a console.

One instance == one layout block.

What a layout contains:

optional title;
one or more status renderers;
one or more block renderers;
a local log buffer;
finish policy: “keep” or “remove”.

Renderer contract remains backward-compatible:

status renderer -> one logical line (str)
block renderer -> multiple logical lines (List[str])

Examples

Create a layout through its owning console:

layout = console.create_layout(title="Download", max_logs=20)

Initialize renderer registries, log storage, and lifecycle state.

Parameters

Name	Type	Description
console	Console	Owning console instance.
title	Optional[str]	Optional heading rendered before layout content.
max_logs	Optional[int]	Maximum retained local log lines, or `None` for no limit.
on_finish	str	`"keep"` retains the finished block; `"remove"` unregisters it. Unknown values are normalized to `"keep"`.
parent	Optional['ConsoleLayout']	Optional parent layout for nested rendering and lifecycle.
indent_step	int	Additional spaces rendered for each nesting level.

Examples

Construct a removable layout directly:

layout = ConsoleLayout(console, title="Task", on_finish="remove")

add_status_renderer(name: str, renderer: Callable[[], str]) → None¶

Parameters

Name	Type	Description
name	str	Stable registry key.
renderer	Callable[[], str]	Zero-argument callable returning one logical line.

Examples

Publish a dynamically evaluated status line:

layout.add_status_renderer("state", lambda: "Ready")

remove_status_renderer(name: str) → None¶

Remove a status renderer by name.

Parameters

Name	Type	Description
name	str	Registry key to remove; missing keys are ignored.

Examples

Stop rendering a previously registered status:

layout.remove_status_renderer("state")

add_block_renderer(name: str, renderer: Callable[[], List[str]]) → None¶

Parameters

Name	Type	Description
name	str	Stable registry key.
renderer	Callable[[], List[str]]	Zero-argument callable returning logical output lines.

Examples

Publish a dynamic two-line details block:

layout.add_block_renderer("details", lambda: ["Files: 3", "Errors: 0"])

remove_block_renderer(name: str) → None¶

Remove a block renderer by name.

Parameters

Name	Type	Description
name	str	Registry key to remove; missing keys are ignored.

Examples

Stop rendering a details block:

layout.remove_block_renderer("details")

append_log(text: str) → None¶

Append a local log line to this layout block.

Parameters

Name	Type	Description
text	str	Value converted to text and retained according to `max_logs`.

Examples

Add one line beneath the live renderers:

layout.append_log("Downloaded page 4")

clear_logs() → None¶

Clear all local layout logs.

Examples

Remove retained history from one layout:

layout.clear_logs()

begin_batch() → None¶

Begin a deferred layout update batch.

Examples

Group several manual mutations before a matching end_batch:

layout.begin_batch()

end_batch() → None¶

End a deferred layout update batch.

Raises

Exception	Description
RuntimeError	No matching `begin_batch()` is active.

Examples

Release a manually opened update batch:

layout.end_batch()

batch_update()¶

Defer layout refreshes until a managed group of updates completes.

Yields

This layout instance.

Examples

Coalesce multiple renderer changes into one refresh:

with layout.batch_update():
    layout.append_log("one")
    layout.append_log("two")

assemble()¶

Atomic assembly helper for layout-bound components.

Suspends both:

layout-local refresh bursts;
console-global redraw bursts.

Yields

This layout instance while local and global redraws are suspended.

Examples

Construct several bound components without partial frames:

with layout.assemble():
    progress = layout.create_progress(10)
    spinner = layout.create_spinner("Waiting")

refresh() → None¶

Ask Console to redraw footer/layout area.

Examples

Repaint after mutating renderer-owned state:

layout.refresh()

finish() → None¶

Mark layout as finished and apply finish policy.

Examples

Complete a layout and retain or remove it according to policy:

layout.finish()

fail(message: str | None = None) → None¶

Mark the layout as failed and apply its finish policy.

Parameters

Name	Type	Description
message	str \| None	Optional failure line appended before finalization.

Examples

Preserve a visible failure summary:

layout.fail("Browser startup failed")

start() → ConsoleLayout¶

Mark the layout active and return it for manual lifecycle control.

Returns

Type	Description
ConsoleLayout	This layout instance.

Examples

Start and later finish a manually managed group:

group = console.group("Import").start()
group.finish()

remove(recursive: bool = True) → None¶

Unregister this layout from Console.

Parameters

Name	Type	Description
recursive	bool	Remove all descendants before detaching this layout.

Examples

Remove one live block immediately:

layout.remove(recursive=True)

show() → None¶

Make this layout and its descendants visible.

Examples

Restore a temporarily hidden group:

layout.show()

hide() → None¶

Hide this layout without destroying its state.

Examples

Temporarily hide a group:

layout.hide()

create_group(title: str | None = None, max_logs: int | None = None, on_finish: str = 'keep', indent_step: int | None = None) → ConsoleLayout¶

Create a child group rendered and removed with this layout.

Parameters

Name	Type	Description
title	str \| None	Optional child heading.
max_logs	int \| None	Maximum retained child log lines.
on_finish	str	`keep` or `remove` child finish policy.
indent_step	int \| None	Child indentation override.

Returns

Type	Description
ConsoleLayout	Registered child layout.

Examples

Nest browser startup inside a project group:

browser = project.create_group("Browser", on_finish="remove")

group(title: str | None = None, max_logs: int | None = None, on_finish: str = 'keep', indent_step: int | None = None) → ConsoleLayout¶

Create a child group rendered and removed with this layout.

Parameters

Name	Type	Description
title	str \| None	Optional child heading.
max_logs	int \| None	Maximum retained child log lines.
on_finish	str	`keep` or `remove` child finish policy.
indent_step	int \| None	Child indentation override.

Returns

Type	Description
ConsoleLayout	Registered child layout.

Examples

Nest browser startup inside a project group:

browser = project.create_group("Browser", on_finish="remove")

get_logger(config=None)¶

Return a layout-local logger bound to this layout.

Parameters

Name	Description
config	Optional layout logger configuration.

Returns

New LayoutLogger attached to this layout.

Examples

Emit records inside one layout:

logger = layout.get_logger()

create_progress(total: int, description: str = 'Progress', **kwargs)¶

Create a ProgressBar bound to this layout.

Supported kwargs include both direct rendering parameters and ProgressConfig-based overrides, including:

config=ProgressConfig(…)
format_str / template
width / indent / colors / chars / visibility flags

Parameters

Name	Type	Description
total	int	Target progress value.
description	str	Label displayed beside the progress indicator.
kwargs		Progress configuration or direct rendering overrides.

Returns

New layout-bound ProgressBar.

Examples

Create a ten-step progress indicator:

progress = layout.create_progress(10, description="Import")

create_spinner(message: str = 'Loading.', **kwargs)¶

Create a Spinner bound to this layout.

Supported kwargs include both direct rendering parameters and SpinnerConfig-based overrides, including:

config=SpinnerConfig(…)
format_str / template
frames / interval / colors / position

Parameters

Name	Type	Description
message	str	Initial spinner message.
kwargs		Spinner configuration or direct rendering overrides.

Returns

New layout-bound spinner.

Examples

Create a spinner sharing this layout:

spinner = layout.create_spinner("Connecting")

create_table(headers=None, rows=None, **kwargs)¶

Create a TableView bound to this layout.

Supported kwargs include both direct rendering parameters and TableConfig-based overrides, including:

config=TableConfig(…)
title / title_style / header_style / border_style
separators / empty-view settings

Parameters

Name	Description
headers	Column headings; `None` creates an empty list.
rows	Initial rows; `None` creates an empty list.
kwargs	Table configuration or direct rendering overrides.

Returns

New layout-bound TableView.

Examples

Create a result table in the live block:

table = layout.create_table(["Case", "Status"], [["42", "Ready"]])

render_lines() → List[str]¶

Render the full layout block to a list of logical visible lines.

These are still plain strings on the public boundary. Console later converts them into Rich Text / Segment lines internally.

Returns

Type	Description
List[str]	Logical ANSI-capable lines, or an empty list when hidden.

Examples

Inspect the current logical frame without writing it:

lines = layout.render_lines()

class ddp_utils.console.console.Console(cursor: ConsoleCursor | None = None, auto_install_proxy: bool = False, intercept_stderr: bool = False, terminal_mode: TerminalMode | str = TerminalMode.AUTO, output_stream: TextIO | None = None, error_stream: TextIO | None = None)¶

Bases: object

Own coordinated terminal output and live-area repainting.

Responsibilities:

coordinated plain writes and print();
stdout/stderr interception;
footer/layout redraw;
standalone live renderer registry;
multi-layout registry.

Internal redraw model:

collect logical footer lines;
convert them to Rich Text / Group;
render physical segment lines via Rich Console.render_lines();
repaint footer by absolute anchor-row addressing.

This removes the old dependency on “footer height guess -> move_up(N)”.

Examples

Manage interception for one bounded operation:

with Console() as console:
    with console.intercepted(intercept_stderr=True):
        print("coordinated output")

Initialize stream ownership, registries, locks, and render state.

Parameters

Name	Type	Description
cursor	Optional[ConsoleCursor]	Cursor controller, or `None` to create a warning-enabled `ConsoleCursor`.
auto_install_proxy	bool	Whether to replace `sys.stdout` immediately.
intercept_stderr	bool	Whether immediate proxy installation also replaces `sys.stderr`.
terminal_mode	TerminalMode \| str	Requested `auto`, `vt`, `win32`, or `plain` rendering mode. Unsafe explicit modes degrade to `plain`.
output_stream	Optional[TextIO]	Output destination captured by this console.
error_stream	Optional[TextIO]	Error destination captured by this console.

Examples

Coordinate both standard streams from construction:

console = Console(auto_install_proxy=True, intercept_stderr=True)

install_stdout_proxy(intercept_stderr: bool = False) → None¶

Install stdout proxy interception.

Parameters

Name	Type	Description
intercept_stderr	bool	Whether to proxy stderr in addition to stdout.

Examples

Coordinate writes made through both standard streams:

console.install_stdout_proxy(intercept_stderr=True)

uninstall_stdout_proxy() → None¶

Uninstall stdout/stderr proxy interception and restore real streams.

Examples

Restore streams captured when the console was created:

console.uninstall_stdout_proxy()

install_print_hook() → None¶

Replace builtins.print with Console.print.

Examples

Coordinate calls that resolve builtins.print dynamically:

console.install_print_hook()

uninstall_print_hook() → None¶

Restore original builtins.print.

Examples

Release process-wide print ownership:

console.uninstall_print_hook()

install_interception(intercept_stderr: bool = False, hook_print: bool = True) → None¶

Enable coordinated interception for stdout/stderr and optionally print().

Parameters

Name	Type	Description
intercept_stderr	bool	Whether to proxy stderr as well as stdout.
hook_print	bool	Whether to replace `builtins.print()`.

Examples

Enable the complete interception surface:

console.install_interception(intercept_stderr=True, hook_print=True)

uninstall_interception() → None¶

Disable coordinated interception.

Examples

Restore the original streams and print function:

console.uninstall_interception()

create_layout(title: str | None = None, max_logs: int | None = None, on_finish: str = 'keep', defer_initial_redraw: bool = False, parent: ConsoleLayout | None = None, indent_step: int = 2) → ConsoleLayout¶

Create and register a new independent layout block.

Parameters

Name	Type	Description
title	str \| None	Optional heading for the layout.
max_logs	int \| None	Maximum local history lines, or `None` for no limit.
on_finish	str	`"keep"` or `"remove"` lifecycle policy.
defer_initial_redraw	bool	Register without drawing the incomplete initial state, allowing atomic component assembly.
parent	ConsoleLayout \| None	Optional parent layout. Prefer `parent.create_group()` for naturally nested code.
indent_step	int	Spaces added for each nesting level.

Returns

Type	Description
ConsoleLayout	Registered layout owned by this console.

Examples

Create a bounded log layout removed at completion:

layout = console.create_layout(
    title="Import", max_logs=25, on_finish="remove"
)

group(title: str | None = None, max_logs: int | None = None, on_finish: str = 'keep', parent: ConsoleLayout | None = None, indent_step: int = 2) → ConsoleLayout¶

Create a root or nested lifecycle group.

Parameters

Name	Type	Description
title	str \| None	Optional group heading.
max_logs	int \| None	Maximum retained group log lines.
on_finish	str	`keep` or `remove` lifecycle policy.
parent	ConsoleLayout \| None	Optional parent group.
indent_step	int	Spaces added for each nesting level.

Returns

Type	Description
ConsoleLayout	Registered `ConsoleLayout` usable manually or with `with`.

Examples

Create a temporary nested group:

with console.group("Project", on_finish="remove") as project:
    with project.group("Browser"):
        launch_browser()

register_layout(layout: ConsoleLayout) → None¶

Parameters

Name	Type	Description
layout	ConsoleLayout	Layout to add or replace by its identifier.

Examples

Attach a directly constructed layout:

console.register_layout(layout)

unregister_layout(layout_id: str) → None¶

Remove a layout from registry and redraw footer.

Parameters

Name	Type	Description
layout_id	str	Identifier to remove; unknown identifiers are ignored.

Examples

Remove a completed live block:

console.unregister_layout(layout.layout_id)

register_renderer(name: str, renderer: Callable[[], List[str] | str]) → None¶

Renderer contract:

may return one string;
may return a list of strings;
Console normalizes that into List[str].

Parameters

Name	Type	Description
name	str	Stable renderer registry key.
renderer	Callable[[], List[str] \| str]	Zero-argument callable returning one line or a list.

Examples

Add a live status outside any layout:

console.register_renderer("status", lambda: "Ready")

unregister_renderer(name: str) → None¶

Remove a standalone live renderer.

Parameters

Name	Type	Description
name	str	Registry key to remove; missing keys are ignored.

Examples

Stop rendering a standalone status:

console.unregister_renderer("status")

push_service_message(text: str) → str¶

Add one temporary message to the live-area stack.

Behavior:

does NOT overwrite previous temporary messages;
appears above footer/layout live blocks;
survives until explicit removal or timeout callback from logger.

Parameters

Name	Type	Description
text	str	Temporary message rendered according to progress placement.

Returns

Type	Description
str	Unique token used to remove this message.

Examples

Display and retain a temporary update:

token = console.push_service_message("Checking updates")

remove_service_message(token: str) → bool¶

Remove one temporary message by token.

Parameters

Name	Type	Description
token	str	Identifier returned by `push_service_message()`.

Returns

Type	Description
bool	`True` when one matching message was removed.

Examples

Remove one temporary update:

removed = console.remove_service_message(token)

clear_service_messages() → bool¶

Remove all temporary messages from the live-area stack.

Returns

Type	Description
bool	`True` when the stack contained at least one message.

Examples

Clear all temporary status output before continuing:

changed = console.clear_service_messages()

has_service_messages() → bool¶

Return True if there are active temporary messages.

Returns

Type	Description
bool	`True` when the service-message stack is non-empty.

Examples

Check whether an ordinary write will consume temporary messages:

pending = console.has_service_messages()

set_active_progress(progress) → None¶

Console uses this only to decide where coordinated ordinary writes must go relative to the live-area.

Parameters

Name	Description
progress	Standalone progress object exposing placement state.

Examples

Track the progress whose placement policy controls writes:

console.set_active_progress(progress)

clear_active_progress(progress=None) → None¶

Clear the currently active standalone progress.

If progress is provided, clear only if it matches the currently tracked active instance. This prevents stale progress objects from wiping a newer one.

Parameters

Name	Description
progress	Optional ownership guard; a different active object is kept.

Examples

Clear only if this progress still owns the active slot:

console.clear_active_progress(progress)

request_redraw() → None¶

Request a footer redraw with throttling/coalescing.

Examples

Mark live output dirty after component state changes:

console.request_redraw()

redraw() → None¶

Request repaint through the public compatibility entry point.

Examples

Refresh all registered live components:

console.redraw()

suspend_redraw()¶

Temporarily suspend footer redraw.

This is used for atomic initial layout assembly.

Yields

This console while repaint requests are accumulated.

Raises

Exception	Description
RuntimeError	Internal suspension depth underflows.

Examples

Assemble several visible components atomically:

with console.suspend_redraw():
    first = console.create_layout(defer_initial_redraw=True)
    second = console.create_layout(defer_initial_redraw=True)

write(text: str, stream: TextIO | None = None) → None¶

Coordinated raw text write.

Parameters

Name	Type	Description
text	str	Value converted to text before writing.
stream	TextIO \| None	Destination, defaulting to captured real stdout.

Examples

Write history without corrupting a live progress frame:

console.write("Downloaded page 4\n")

print(*args, sep: str = ' ', end: str = '\n', file: TextIO | None = None, flush: bool = True) → None¶

Coordinated print compatible with builtins.print.

Parameters

Name	Type	Description
args		Values joined after conversion to text; `None` becomes empty.
sep	str	Separator placed between values.
end	str	Text appended after the joined values.
file	TextIO \| None	Destination, defaulting to captured real stdout.
flush	bool	Whether to flush the destination after the coordinated write.

Examples

Print several values using the coordinated path:

console.print("page", 4, sep="=")

cprint(text: str, color: str | None = None, bg: str | None = None, style=None, end: str = '\n') → None¶

Coordinated styled print using styler-prepared ANSI text.

Parameters

Name	Type	Description
text	str	Text to style and write.
color	str \| None	Optional foreground color name.
bg	str \| None	Optional background color name.
style		Optional style name or collection accepted by `colorize`.
end	str	Text appended after the styled value.

Examples

Emit a successful status line in green:

console.cprint("Ready", color="green", style="bold")

hold()¶

Manual lock hold for complex atomic terminal sequences.

Yields

This console while its reentrant lock is held.

Examples

Serialize a custom sequence with coordinated output:

with console.hold():
    console.write("first\n")
    console.write("second\n")

intercepted(intercept_stderr: bool = False, hook_print: bool = True)¶

Temporarily enable stdout/stderr/print interception.

Parameters

Name	Type	Description
intercept_stderr	bool	Whether to intercept stderr as well as stdout.
hook_print	bool	Whether to replace `builtins.print()`.

Yields

This console with requested interception installed.

Examples

Coordinate ordinary print calls for one operation:

with console.intercepted():
    print("safe beside progress")

close() → None¶

Cancel repainting, clear live state, and restore process streams.

Examples

Release console ownership explicitly:

console.close()

ddp_utils.console.console.get_console() → Console¶

Return the shared Console singleton.

Returns

Type	Description
Console	Process-local lazily created console instance.

Examples

Share one coordinator between console components:

console = get_console()