qq_lib.queues
Presentation utilities for queues of the batch system.
This module defines QueuesPresenter, a formatter that turns raw queue data from
the batch system into user-friendly Rich panels. It summarizes per-queue load,
availability, routing relationships, limits such as walltime and node caps, and
optional administrative comments.
1# Released under MIT License. 2# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab 3 4""" 5Presentation utilities for queues of the batch system. 6 7This module defines `QueuesPresenter`, a formatter that turns raw queue data from 8the batch system into user-friendly Rich panels. It summarizes per-queue load, 9availability, routing relationships, limits such as walltime and node caps, and 10optional administrative comments. 11""" 12 13from .presenter import QueuesPresenter 14 15__all__ = [ 16 "QueuesPresenter", 17]
class
QueuesPresenter:
16class QueuesPresenter: 17 """ 18 Presents information about queues of the batch system. 19 """ 20 21 def __init__( 22 self, 23 queues: list[BatchQueueInterface], 24 user: str, 25 all: bool, 26 server: str | None, 27 ): 28 """ 29 Initialize the presenter with a list of queues. 30 31 Args: 32 queues (list[BatchQueueInterface]): List of queue information objects 33 to be presented. 34 user (str): Name of the user for which the queues are displayed. 35 all (bool): Display all queues or only those available to the user. 36 server (str | None): Batch server for which the queues were collected. 37 `None` = default server. 38 """ 39 self._queues = queues 40 self._user = user 41 self._display_all = all 42 self._server = server 43 44 self._show_comment = self._should_show_comment() 45 self._show_max_nnodes = self._should_show_max_n_nodes() 46 47 def dump_yaml(self) -> None: 48 """ 49 Print the YAML representation of all queues to stdout. 50 """ 51 for queue in self._queues: 52 print(queue.to_yaml()) 53 54 def create_queues_info_panel(self, console: Console | None = None) -> Group: 55 """ 56 Create a Rich panel displaying queue information. 57 58 Args: 59 console (Console | None): Optional Rich Console instance. 60 If None, a new Console will be created. 61 62 Returns: 63 Group: Rich Group containing the queues table. 64 """ 65 console = console or Console() 66 queues_table = self._create_queues_table() 67 68 panel = Panel( 69 queues_table, 70 title=Text( 71 f"{'ALL' if self._display_all else 'AVAILABLE'} QUEUES", 72 style=CFG.queues_presenter.title_style, 73 justify="center", 74 ), 75 subtitle=Text( 76 f"{self._server}", 77 style=CFG.jobs_presenter.subtitle_style, 78 justify="center", 79 ) 80 if self._server 81 else None, 82 border_style=CFG.queues_presenter.border_style, 83 padding=(1, 1), 84 width=get_panel_width( 85 console, 86 1, 87 CFG.queues_presenter.min_width, 88 CFG.queues_presenter.max_width, 89 ), 90 expand=False, 91 ) 92 93 return Group(Text(""), panel, Text("")) 94 95 def _create_queues_table(self) -> Table: 96 """ 97 Construct and return a formatted Rich Table containing queue information. 98 99 Returns: 100 Table: A Rich Table object populated with formatted queue data. 101 """ 102 table = Table( 103 show_header=True, 104 box=None, 105 padding=(0, 1), 106 ) 107 108 table.add_column(justify="left") 109 table.add_column( 110 header=Text( 111 "Name", justify="center", style=CFG.queues_presenter.headers_style 112 ), 113 justify="left", 114 ) 115 table.add_column( 116 header=Text( 117 "Priority", justify="center", style=CFG.queues_presenter.headers_style 118 ), 119 justify="center", 120 ) 121 table.add_column( 122 header=Text( 123 BatchState.RUNNING.to_code(), 124 justify="right", 125 style=CFG.state_colors.running, 126 ), 127 justify="right", 128 ) 129 table.add_column( 130 header=Text( 131 f"{BatchState.QUEUED.to_code()}{BatchState.HELD.to_code()}", 132 justify="right", 133 style=CFG.state_colors.queued, 134 ), 135 justify="right", 136 ) 137 table.add_column( 138 header=Text( 139 CFG.queues_presenter.other_jobs_code, 140 justify="right", 141 style=CFG.state_colors.other, 142 ), 143 justify="right", 144 ) 145 table.add_column( 146 header=Text( 147 CFG.queues_presenter.sum_jobs_code, 148 justify="right", 149 style=CFG.state_colors.sum, 150 ), 151 justify="right", 152 ) 153 table.add_column( 154 header=Text( 155 "Max Walltime", 156 justify="center", 157 style=CFG.queues_presenter.headers_style, 158 ), 159 justify="right", 160 ) 161 if self._show_max_nnodes: 162 table.add_column( 163 header=Text( 164 "Max Nodes", 165 justify="center", 166 style=CFG.queues_presenter.headers_style, 167 ), 168 justify="center", 169 ) 170 171 if self._show_comment: 172 table.add_column( 173 header=Text( 174 "Comment", 175 justify="center", 176 style=CFG.queues_presenter.headers_style, 177 ), 178 justify="center", 179 ) 180 181 visited_queues = set() 182 for queue in self._queues: 183 if queue.from_route_only(): 184 continue 185 186 self._add_queue_row(queue, table, self._user) 187 visited_queues.add(queue) 188 189 # print all reroutings 190 if dest_names := queue.get_destinations(): 191 destinations = [q for q in self._queues if q.get_name() in dest_names] 192 193 for rerouted in destinations: 194 self._add_queue_row( 195 rerouted, 196 table, 197 self._user, 198 from_route=True, 199 # we set the availability to False if the parent queue is not available 200 available=None 201 if queue.is_available_to_user(self._user) 202 else False, 203 ) 204 visited_queues.add(rerouted) 205 206 # print all unbound queues (from route only queues that do not have a parent) 207 if self._display_all and ( 208 unvisited_queues := set(self._queues) - visited_queues 209 ): 210 table.add_row( 211 Text("?", style=f"{CFG.queues_presenter.dangling_mark_style} bold") 212 ) 213 for queue in unvisited_queues: 214 self._add_queue_row( 215 queue, table, self._user, from_route=True, dangling=True 216 ) 217 218 return table 219 220 def _add_queue_row( 221 self, 222 queue: BatchQueueInterface, 223 table: Table, 224 user: str, 225 from_route: bool = False, 226 dangling: bool = False, 227 available: bool | None = None, 228 ): 229 """ 230 Add a formatted row representing a single queue to the given table. 231 232 Args: 233 queue (BatchQueueInterface): The queue to display. 234 table (Table): The Rich Table instance to modify. 235 user (str): The username used to determine queue availability. 236 from_route (bool, optional): Indicates whether the queue is a rerouted destination. 237 Defaults to False. 238 dangling (bool, optional): Marks the queue as unbound destination (no parent route). Defaults to False. 239 available (bool | None, optional): Manually overrides queue availability. If None, 240 availability is determined automatically. Defaults to None. 241 """ 242 mark = ( 243 CFG.queues_presenter.main_mark 244 if not from_route 245 else CFG.queues_presenter.rerouted_mark 246 ) 247 248 available = available or queue.is_available_to_user(user) 249 250 if available and dangling: 251 mark_style = CFG.queues_presenter.dangling_mark_style 252 elif available: 253 mark_style = CFG.queues_presenter.available_mark_style 254 else: 255 mark_style = CFG.queues_presenter.unavailable_mark_style 256 257 text_style = ( 258 CFG.queues_presenter.main_text_style 259 if not from_route 260 else CFG.queues_presenter.rerouted_text_style 261 ) 262 263 content = [ 264 Text(mark, style=mark_style), 265 Text(queue.get_name(), style=text_style), 266 Text(queue.get_priority() or "", style=text_style), 267 Text(str(queue.get_running_jobs() or 0), style=CFG.state_colors.running), 268 Text(str(queue.get_queued_jobs() or 0), style=CFG.state_colors.queued), 269 Text(str(queue.get_other_jobs() or 0), style=CFG.state_colors.other), 270 Text(str(queue.get_total_jobs() or 0), style=CFG.state_colors.sum), 271 QueuesPresenter._format_walltime(queue, text_style), 272 Text(str(queue.get_max_n_nodes() or "∞"), style=text_style) 273 if self._show_max_nnodes 274 else None, 275 Text(queue.get_comment() or "", style=text_style) 276 if self._show_comment 277 else None, 278 ] 279 280 table.add_row(*[x for x in content if x is not None]) 281 282 @staticmethod 283 def _format_walltime(queue: BatchQueueInterface, style: str) -> Text: 284 """ 285 Format the queue's maximum walltime for display. 286 287 Args: 288 queue (BatchQueueInterface): The queue whose walltime is being formatted. 289 style (str): The Rich text style to apply. 290 291 Returns: 292 Text: A styled Rich Text object containing the formatted walltime, 293 or an empty string if no walltime is defined. 294 """ 295 if not (walltime := queue.get_max_walltime()): 296 return Text("") 297 298 return Text(format_duration_wdhhmmss(walltime), style=style) 299 300 def _should_show_comment(self) -> bool: 301 """ 302 Determine whether the Comment column should be displayed. 303 304 Returns: 305 bool: True if any node has a comment, False otherwise. 306 """ 307 return any(queue.get_comment() is not None for queue in self._queues) 308 309 def _should_show_max_n_nodes(self) -> bool: 310 """ 311 Determine whether the Max Nodes column should be displayed. 312 313 Returns: 314 bool: True if any node has a defined maximal number of nodes that 315 can be requested, False otherwise. 316 """ 317 return any(queue.get_max_n_nodes() is not None for queue in self._queues)
Presents information about queues of the batch system.
QueuesPresenter( queues: list[qq_lib.batch.interface.BatchQueueInterface], user: str, all: bool, server: str | None)
21 def __init__( 22 self, 23 queues: list[BatchQueueInterface], 24 user: str, 25 all: bool, 26 server: str | None, 27 ): 28 """ 29 Initialize the presenter with a list of queues. 30 31 Args: 32 queues (list[BatchQueueInterface]): List of queue information objects 33 to be presented. 34 user (str): Name of the user for which the queues are displayed. 35 all (bool): Display all queues or only those available to the user. 36 server (str | None): Batch server for which the queues were collected. 37 `None` = default server. 38 """ 39 self._queues = queues 40 self._user = user 41 self._display_all = all 42 self._server = server 43 44 self._show_comment = self._should_show_comment() 45 self._show_max_nnodes = self._should_show_max_n_nodes()
Initialize the presenter with a list of queues.
Arguments:
- queues (list[BatchQueueInterface]): List of queue information objects to be presented.
- user (str): Name of the user for which the queues are displayed.
- all (bool): Display all queues or only those available to the user.
- server (str | None): Batch server for which the queues were collected.
None= default server.
def
dump_yaml(self) -> None:
47 def dump_yaml(self) -> None: 48 """ 49 Print the YAML representation of all queues to stdout. 50 """ 51 for queue in self._queues: 52 print(queue.to_yaml())
Print the YAML representation of all queues to stdout.
def
create_queues_info_panel(self, console: rich.console.Console | None = None) -> rich.console.Group:
54 def create_queues_info_panel(self, console: Console | None = None) -> Group: 55 """ 56 Create a Rich panel displaying queue information. 57 58 Args: 59 console (Console | None): Optional Rich Console instance. 60 If None, a new Console will be created. 61 62 Returns: 63 Group: Rich Group containing the queues table. 64 """ 65 console = console or Console() 66 queues_table = self._create_queues_table() 67 68 panel = Panel( 69 queues_table, 70 title=Text( 71 f"{'ALL' if self._display_all else 'AVAILABLE'} QUEUES", 72 style=CFG.queues_presenter.title_style, 73 justify="center", 74 ), 75 subtitle=Text( 76 f"{self._server}", 77 style=CFG.jobs_presenter.subtitle_style, 78 justify="center", 79 ) 80 if self._server 81 else None, 82 border_style=CFG.queues_presenter.border_style, 83 padding=(1, 1), 84 width=get_panel_width( 85 console, 86 1, 87 CFG.queues_presenter.min_width, 88 CFG.queues_presenter.max_width, 89 ), 90 expand=False, 91 ) 92 93 return Group(Text(""), panel, Text(""))
Create a Rich panel displaying queue information.
Arguments:
- console (Console | None): Optional Rich Console instance. If None, a new Console will be created.
Returns:
Group: Rich Group containing the queues table.