CursesMonitor

A curses based Celery task monitor.

Attributes

Attribute	Type	Description
keymap	`dict` = {}	Mapping of keyboard characters to handler methods used for processing user input.
win	`window` = null	The curses window object used for all screen drawing and input operations.
screen_delay	`int` = 10	The duration in milliseconds to pause between screen updates during the nap cycle.
selected_task	`string` = null	The UUID of the currently highlighted task in the monitor interface.
selected_position	`int` = 0	The numerical index of the currently selected row in the task list.
selected_str	`string` = Selected:	The prefix label displayed in the footer when showing details for a selected task.
foreground	`int` = curses.COLOR_BLACK	The curses color constant used for the primary text color of the interface.
background	`int` = curses.COLOR_WHITE	The curses color constant used for the background color of the interface.
online_str	`string` = Workers online:	The label used in the footer to introduce the list of active workers.
help_title	`string` = Keys:	The label used to introduce the keyboard shortcut help section in the footer.
help	`string` = j:down k:up i:info t:traceback r:result c:revoke ^c: quit	Verbatim string describing available key bindings shown at the bottom of the screen.
greet	`string` = celery events {VERSION_BANNER}	The greeting message and version banner displayed at the top of the monitor.
info_str	`string` = Info:	The label used to introduce the general status information line in the footer.
keyalias	`dict` = {curses.KEY_DOWN: 'J', curses.KEY_UP: 'K', curses.KEY_ENTER: 'I'}	Mapping of curses key constants to their equivalent character representations for unified input handling.

Constructor

Signature

def CursesMonitor(
    state: Any,
    app: [Celery](../../app/base/celery.md?sid=celery_app_base_celery),
    keymap: dict = None
) - > null

Parameters

Name	Type	Description
state	`Any`	The current state of the Celery workers and tasks.
app	`[Celery](../../app/base/celery.md?sid=celery_app_base_celery)`	The Celery application instance.
keymap	`dict` = None	Optional dictionary mapping keys to handler methods to override or extend default behavior.

Methods

`format_row()`

@classmethod
def format_row(
    uuid: string,
    task: string,
    worker: string,
    timestamp: string,
    state: string
) - > string

Formats a single task row for display, calculating column widths based on the current screen dimensions.

Parameters

Name	Type	Description
uuid	`string`	The unique identifier of the task.
task	`string`	The name of the task.
worker	`string`	The hostname of the worker executing the task.
timestamp	`string`	The formatted time string for the task event.
state	`string`	The current execution state of the task.

Returns

Type	Description
`string`	A formatted string containing the task UUID, worker, name, timestamp, and state, truncated to fit the display width.

`screen_width()`

@classmethod
def screen_width() - > integer

Retrieves the total width of the current curses window.

Returns

Type	Description
`integer`	The maximum number of columns available in the window.

`screen_height()`

@classmethod
def screen_height() - > integer

Retrieves the total height of the current curses window.

Returns

Type	Description
`integer`	The maximum number of rows available in the window.

`display_width()`

@classmethod
def display_width() - > integer

Calculates the usable width for task data, accounting for border spacing.

Returns

Type	Description
`integer`	The width available for content after subtracting border offsets.

`display_height()`

@classmethod
def display_height() - > integer

Calculates the usable height for the task list, reserving space for headers and footers.

Returns

Type	Description
`integer`	The number of rows available to display task entries.

`limit()`

@classmethod
def limit() - > integer

Determines the maximum number of tasks that can be displayed on the current screen.

Returns

Type	Description
`integer`	The display height limit for the task list.

`find_position()`

@classmethod
def find_position() - > integer

Locates the index of the currently selected task within the visible task list.

Returns

Type	Description
`integer`	The zero-based index of the selected task, or 0 if no task is selected or found.

`move_selection_up()`

@classmethod
def move_selection_up()

Moves the UI selection highlight to the previous task in the list.

`move_selection_down()`

@classmethod
def move_selection_down()

Moves the UI selection highlight to the next task in the list.

`move_selection()`

@classmethod
def move_selection(
    direction: integer
)

Updates the selected task by shifting the current position by a specified direction.

Parameters

Name	Type	Description
direction	`integer`	The number of positions to move; positive for down, negative for up.

`handle_keypress()`

@classmethod
def handle_keypress()

Reads a key from the curses window and executes the corresponding command from the keymap.

`alert()`

@classmethod
def alert(
    callback: callable,
    title: string
) - > string

Displays a temporary modal-like screen with a title and custom content, waiting for a keypress to dismiss.

Parameters

Name	Type	Description
callback	`callable`	A function used to render the body of the alert; receives max_y, max_x, and starting y-coordinate.
title	`string`	The header text displayed at the top of the alert screen.

Returns

Type	Description
`string`	The uppercase string representation of the key pressed to exit the alert.

`selection_rate_limit()`

@classmethod
def selection_rate_limit()

Prompts the user for a new rate limit and applies it to the currently selected task's type.

`alert_remote_control_reply()`

@classmethod
def alert_remote_control_reply(
    reply: list
) - > string

Displays an alert showing the responses received from workers after a remote control command.

Parameters

Name	Type	Description
reply	`list`	A list of dictionaries containing worker hostnames and their respective command responses.

Returns

Type	Description
`string`	The key pressed to dismiss the alert.

`readline()`

@classmethod
def readline(
    x: integer,
    y: integer
) - > string

Captures a line of text input from the user at a specific screen location.

Parameters

Name	Type	Description
x	`integer`	The vertical row coordinate where input starts.
y	`integer`	The horizontal column coordinate where input starts.

Returns

Type	Description
`string`	The string entered by the user, or an empty string if cancelled via Escape.

`revoke_selection()`

@classmethod
def revoke_selection()

Sends a revoke command to the cluster for the currently selected task.

`selection_info()`

@classmethod
def selection_info()

Displays a detailed view of the selected task's metadata, including arguments and keyword arguments.

`selection_traceback()`

@classmethod
def selection_traceback()

Displays the error traceback for the selected task if it failed with an exception.

`selection_result()`

@classmethod
def selection_result()

Displays the return value or exception message of the currently selected task.

`display_task_row()`

@classmethod
def display_task_row(
    lineno: integer,
    task: object
)

Renders a single task's information onto a specific line of the curses window with appropriate styling.

Parameters

Name	Type	Description
lineno	`integer`	The row index in the window where the task should be drawn.
task	`object`	The task state object containing UUID, name, worker, and status info.

`draw()`

@classmethod
def draw()

Refreshes the entire monitor UI, including the task list, selection details, worker status, and help footer.

`safe_add_str()`

@classmethod
def safe_add_str(
    y: integer,
    x: integer,
    string: string
)

Adds a string to the window while ensuring it does not exceed the screen width to prevent curses errors.

Parameters

Name	Type	Description
y	`integer`	The row coordinate.
x	`integer`	The column coordinate.
string	`string`	The text to display.

`init_screen()`

@classmethod
def init_screen()

Initializes the curses environment, sets up colors, and configures input modes.

`resetscreen()`

@classmethod
def resetscreen()

Restores the terminal to its original state and closes the curses window.

`nap()`

@classmethod
def nap()

Suspends execution for a short duration defined by the screen delay to control refresh frequency.

`tasks()`

@classmethod
def tasks() - > list

Retrieves the current list of tasks from the state, limited to what can fit on the screen.

Returns

Type	Description
`list`	A list of task tuples sorted by time.

`workers()`

@classmethod
def workers() - > list

Retrieves a list of hostnames for all workers currently marked as alive.

Returns

Type	Description
`list`	A list of strings representing active worker hostnames.

Attributes​

Constructor​

Signature​

Parameters​

Methods​

format_row()​

Parameters​

Returns​

screen_width()​

Returns​

screen_height()​

Returns​

display_width()​

Returns​

display_height()​

Returns​

limit()​

Returns​

find_position()​

Returns​

move_selection_up()​

move_selection_down()​

move_selection()​

Parameters​

handle_keypress()​

alert()​

Parameters​

Returns​

selection_rate_limit()​

alert_remote_control_reply()​

Parameters​

Returns​

readline()​

Parameters​

Returns​

revoke_selection()​

selection_info()​

selection_traceback()​

selection_result()​

display_task_row()​

Parameters​

draw()​

safe_add_str()​

Parameters​

init_screen()​

resetscreen()​

nap()​

tasks()​

Returns​

workers()​

Returns​

Attributes

Constructor

Signature

Parameters

Methods

`format_row()`

Parameters

Returns

`screen_width()`

Returns

`screen_height()`

Returns

`display_width()`

Returns

`display_height()`

Returns

`limit()`

Returns

`find_position()`

Returns

`move_selection_up()`

`move_selection_down()`

`move_selection()`

Parameters

`handle_keypress()`

`alert()`

Parameters

Returns

`selection_rate_limit()`

`alert_remote_control_reply()`

Parameters

Returns

`readline()`

Parameters

Returns

`revoke_selection()`

`selection_info()`

`selection_traceback()`

`selection_result()`

`display_task_row()`

Parameters

`draw()`

`safe_add_str()`

Parameters

`init_screen()`

`resetscreen()`

`nap()`

`tasks()`

Returns

`workers()`

Returns