# Copyright 2023-2026 Tom Meltzer. See the top-level COPYRIGHT file for
# details.
import re
from os.path import expanduser
from typing import TYPE_CHECKING
from collections import defaultdict
if TYPE_CHECKING:
from .backend import DebugBackend
[docs]
def sort_debug_response(results: dict[int, str]) -> dict[int, str]:
"""Sort debug output by process rank.
Args:
response: dict containing the
Returns:
The string with bracketted paste escape sequence removed.
"""
return dict(sorted(results.items()))
[docs]
def collapse_ranges(values: list[int]) -> str:
"""Collapse a list of integers into minimal range notation.
e.g. [1, 2, 3, 6, 7, 10, 11, 12] -> '1-3,6-7,10-12'
"""
sorted_vals = sorted(set(values))
if not sorted_vals:
return ""
ranges = []
start = end = sorted_vals[0]
for n in sorted_vals[1:]:
if n == end + 1:
end = n
else:
ranges.append(f"{start}-{end}" if start != end else str(start))
start = end = n
ranges.append(f"{start}-{end}" if start != end else str(start))
return ",".join(ranges)
[docs]
def pretty_print_response(response: dict[int, str]) -> None:
lines = []
for rank, result in response.items():
if result:
lines.append(prepend_ranks(rank=rank, result=result))
combined_output = (72 * "*" + "\n").join(lines)
print(combined_output)
[docs]
def reduce_response(response: dict[int, str]) -> None:
"""Reduce debug output by deduplicating common lines across ranks.
Parses each rank's output, groups identical lines together, and prints
each unique line prefixed by the collapsed set of ranks that produced it.
Lines from only one rank appear normally; shared lines are printed once
with all contributing ranks.
Example output before and after:
Before (raw output per rank):
0: process 45402
0: cmdline = '/mdb/examples/simple-mpi.exe'
0: cwd = '/mdb'
0: exe = '/mdb/examples/simple-mpi.exe'
************************************************************************
1: process 45403
1: cmdline = '/mdb/examples/simple-mpi.exe'
1: cwd = '/mdb'
1: exe = '/mdb/examples/simple-mpi.exe'
After (deduplicated, ranks collapsed):
0: process 45402
0-1: cmdline = '/mdb/examples/simple-mpi.exe'
0-1: cwd = '/mdb'
0-1: exe = '/mdb/examples/simple-mpi.exe'
1: process 45403
Args:
response: dict mapping process rank (int) to its output string.
"""
reduced = defaultdict(list)
for rank, result in response.items():
if result:
for line in result.split("\r\n")[1:-1]:
reduced[line].append(rank)
reduced = {k: collapse_ranges(v) for k, v in reduced.items()}
max_len = max([len(v) for v in reduced.values()], default=0)
for line, ranks_str in reduced.items():
padded = ranks_str.rjust(max_len)
print(f"{padded}: {line}")
[docs]
def prepend_ranks(rank: int, result: str) -> str:
return "".join([f"{rank}: " + line + "\r\n" for line in result.split("\r\n")[1:-1]])
[docs]
def strip_bracketted_paste(text: str) -> str:
"""Strip bracketted paste escape sequence from text
(see issue #669 https://github.com/pexpect/pexpect/issues/669).
Args:
text: string that contains bracketted paste escape sequence,
Returns:
The string with bracketted paste escape sequence removed.
"""
return re.sub(r"\x1b\[\?2004[lh]\r*", "", text)
[docs]
def strip_control_characters(text: str) -> str:
"""Strip ANSI control characters from a string.
Args:
text: string that contains ANSI control characters.
Returns:
The string with ANSI control characters removed.
"""
return re.sub(r"\x1b\[[\d;]*m", "", text)
[docs]
def parse_ranks(ranks: str) -> list[int]:
"""Parse a string of ranks into a set of integers. E.g.,
``parse_ranks("1,3-5")`` would return the following: ``set(1,3,4,5)``.
Args:
ranks: string of ranks using either a mix of comma separation and
ranges using hyphen.
Returns:
A set of ranks represented by integers.
"""
ranks = re.sub(
r"(\d+)-(\d+)",
lambda match: ",".join(
str(i) for i in range(int(match.group(1)), int(match.group(2)) + 1)
),
ranks,
)
return list(set([int(s) for s in ranks.split(",")]))
[docs]
def ssl_cert_path() -> str:
return expanduser("~/.mdb/cert.pem")
[docs]
def ssl_key_path() -> str:
return expanduser("~/.mdb/key.rsa")