Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 120 additions & 13 deletions arc/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -305,6 +305,66 @@ def log_footer(execution_time: str,
logger.log(level, f'ARC execution terminated on {time.asctime()}')


def format_table(headers: Sequence[str | Sequence[str]],
rows: Sequence[Sequence[str]],
alignments: str | None = None,
separator: str = ' ',
rule_char: str = '-',
) -> list[str]:
"""
Format a table as a list of lines with aligned columns.

Each column is exactly as wide as its widest entry, considering both its header and its cells.
Widths are counted in characters, so the columns line up for single-width text.
A header may be given as a sequence of strings to render it on several lines, e.g., a title
line and a units line; shorter headers are padded with blank lines at the bottom, placing every
header's first string on the first line. Trailing whitespace is stripped from every line.

Args:
headers (Sequence[str | Sequence[str]]): The column headers, each a string,
or a sequence of strings for a multi-line header.
rows (Sequence[Sequence[str]]): The table rows, each a sequence of one cell string per column.
alignments (str, optional): One alignment character ('<', '>' or '^') per column.
Columns are left-aligned if not given.
separator (str, optional): The string separating two adjacent columns.
rule_char (str, optional): The character to draw the rule under the header with,
an empty string to omit the rule.

Returns: list[str]
The rendered lines: the header line(s), a rule, and one line per row.

Raises:
InputError: If a row, or the alignments, does not have exactly one entry per column,
if an alignment character is not one of '<', '>' and '^',
or if a cell is not a string.
"""
headers = [(header,) if isinstance(header, str) else tuple(header) for header in headers]
alignments = alignments if alignments is not None else '<' * len(headers)
if len(alignments) != len(headers):
raise InputError(f'Expected {len(headers)} alignment characters, got {len(alignments)}: {alignments}')
if any(alignment not in '<>^' for alignment in alignments):
raise InputError(f'Alignment characters must each be one of "<", ">" and "^", got {alignments}')
for row in rows:
if len(row) != len(headers):
raise InputError(f'Expected {len(headers)} cells per row, got {len(row)}: {row}')
if any(not isinstance(cell, str) for cell in row):
raise InputError(f'Table cells must be strings, got {row}')
header_height = max((len(header) for header in headers), default=0)
headers = [header + ('',) * (header_height - len(header)) for header in headers]
widths = [max([len(line) for line in header] + [len(row[i]) for row in rows] + [0])
for i, header in enumerate(headers)]

def render(cells: Sequence[str]) -> str:
"""Render one line of cells, padded to the column widths and stripped of trailing space."""
return separator.join(f'{cell:{alignment}{width}}'
for cell, alignment, width in zip(cells, alignments, widths)).rstrip()

lines = [render([header[i] for header in headers]) for i in range(header_height)]
if rule_char:
lines.append(render([rule_char * width for width in widths]))
return lines + [render(row) for row in rows]


def get_git_commit(path: str | None = None) -> tuple[str, str]:
"""
Get the recent git commit to be logged.
Expand Down Expand Up @@ -1368,12 +1428,44 @@ def time_lapse(t0) -> str:
h, m = divmod(m, 60)
d, h = divmod(h, 24)
if d > 0:
d = str(d) + ' days, '
d = f'{d:.0f} days, '
else:
d = ''
return f'{d}{h:02.0f}:{m:02.0f}:{s:02.0f}'


def format_duration(duration: 'datetime.timedelta | str | None') -> str:
Comment thread
github-advanced-security[bot] marked this conversation as resolved.
Fixed
"""
Format a duration briefly, in the largest unit it fills, e.g. '3.4 s', '47.2 m', '13.1 h', '2.1 d'.

Seconds ('s') are used below a minute, minutes ('m') below an hour, hours ('h') below a day,
and days ('d') beyond that. The value is given to one decimal place and is followed by a space
and the symbol of its unit. A duration of exactly zero is reported as '0.0 s'. A negative
duration, and a value that cannot be read as a duration, are reported as an empty string.

A string duration is read by :func:`timedelta_from_str`, so both accept the same grammars.
An empty or blank string is treated as an absent duration and is not reported as unreadable.

Args:
duration (datetime.timedelta, str, optional): The duration, either a timedelta object
or its ``str()`` representation,
e.g. '2 days, 3:04:05.678'.

Returns: str
The brief representation of the duration.
"""
if isinstance(duration, str):
duration = timedelta_from_str(duration) if duration.strip() else None
if not isinstance(duration, datetime.timedelta) or duration.total_seconds() < 0:
return ''
seconds = duration.total_seconds()
for limit, divisor, unit in ((60, 1, 's'), (60, 60, 'm'), (24, 3600, 'h')):
value = seconds / divisor
if round(value, 1) < limit:
return f'{value:.1f} {unit}'
return f'{seconds / 86400:.1f} d'


def estimate_orca_mem_cpu_requirement(num_heavy_atoms: int,
server: str = '',
consider_server_limits: bool = False,
Expand Down Expand Up @@ -1715,26 +1807,41 @@ def get_close_tuple(key_1: tuple[float | str, ...],
raise ValueError(f'Could not locate a key close to {key_1} within the tolerance {tolerance} in the given keys list.')


def timedelta_from_str(time_str: str):
TIMEDELTA_STR_REGEX = re.compile(r'^(?:(?P<days>[-+]?\d+)\s+days?,\s*)?'
r'(?P<hours>\d{1,2}):(?P<minutes>\d{2}):(?P<seconds>\d{2})'
r'(?:\.(?P<microseconds>\d{1,6}))?$')
TIMEDELTA_COMPACT_REGEX = re.compile(r'^(?:(?P<hours>\d+)hr)?(?:(?P<minutes>\d+)m)?(?:(?P<seconds>\d+)s)?$')


def timedelta_from_str(time_str: str) -> datetime.timedelta | None:
"""
Get a datetime.timedelta object from its str() representation
Get a datetime.timedelta object from its string representation.

Two grammars are accepted. The primary one is the ``str(datetime.timedelta)`` representation,
e.g., '0:00:03.420000', '2 days, 3:00:00', or '-1 day, 23:59:59', which is the form ARC persists
in restart.yml and output.yml. The secondary one is a compact 'hr'/'m'/'s' form, e.g., '1hr2m3s'
or '45s', in which at least one of the three components must be present.

Args:
time_str (str): The string representation of a datetime.timedelta object.

Returns:
datetime.timedelta: The corresponding timedelta object.
datetime.timedelta | None: The corresponding timedelta object,
or ``None`` if ``time_str`` does not represent a duration.
"""
regex = re.compile(r'((?P<hours>\d+?)hr)?((?P<minutes>\d+?)m)?((?P<seconds>\d+?)s)?')

parts = regex.match(time_str)
if not parts:
return
parts = parts.groupdict()
if not isinstance(time_str, str) or not time_str.strip():
logger.warning(f'Could not interpret {time_str!r} as a time delta, returning None.')
return None
match = TIMEDELTA_STR_REGEX.match(time_str.strip())
if match is None:
match = TIMEDELTA_COMPACT_REGEX.match(time_str.strip())
if match is None or not any(match.groupdict().values()):
logger.warning(f'Could not interpret {time_str!r} as a time delta, returning None.')
return None
time_params = {}
for (name, param) in parts.items():
if param:
time_params[name] = int(param)
for name, param in match.groupdict().items():
if param is not None:
time_params[name] = int(param.ljust(6, '0')) if name == 'microseconds' else int(param)
return datetime.timedelta(**time_params)


Expand Down
156 changes: 156 additions & 0 deletions arc/common_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,127 @@ def test_time_lapse(self):
lap = common.time_lapse(t0)
self.assertEqual(lap, '00:00:02')

def test_time_lapse_beyond_a_day(self):
"""Test that time_lapse() gives a whole number of days and stays readable by timedelta_from_str()"""
now = time.time()
self.assertEqual(common.time_lapse(now - 30 * 3600), '1 days, 06:00:00')
self.assertEqual(common.time_lapse(now - 3 * 86400), '3 days, 00:00:00')
self.assertEqual(common.time_lapse(now - (400 * 86400 + 1)), '400 days, 00:00:01')
for seconds in [6 * 3600, 23 * 3600 + 59 * 60 + 59, 30 * 3600, 3 * 86400, 400 * 86400 + 1]:
lap = common.time_lapse(now - seconds)
self.assertEqual(common.timedelta_from_str(lap), datetime.timedelta(seconds=seconds),
msg=f'time_lapse() gave {lap!r}, which does not read back as {seconds} s')

def test_format_duration(self):
"""Test the format_duration() function"""
self.assertEqual(common.format_duration(datetime.timedelta(0)), '0.0 s')
self.assertEqual(common.format_duration(datetime.timedelta(seconds=3.42)), '3.4 s')
self.assertEqual(common.format_duration(datetime.timedelta(minutes=7)), '7.0 m')
self.assertEqual(common.format_duration(datetime.timedelta(hours=13, minutes=7)), '13.1 h')
self.assertEqual(common.format_duration(datetime.timedelta(days=2, hours=3)), '2.1 d')
self.assertEqual(common.format_duration(datetime.timedelta(days=140, hours=1)), '140.0 d')

def test_format_duration_keeps_sub_minute_timings_distinct(self):
"""Test that the sub-minute TS guess timings of a real run render to three distinct strings"""
rendered = [common.format_duration(datetime.timedelta(seconds=seconds))
for seconds in [3.4, 16.8, 18.1]]
self.assertEqual(rendered, ['3.4 s', '16.8 s', '18.1 s'])
self.assertEqual(len(set(rendered)), 3)

def test_format_duration_unit_boundaries(self):
"""Test that format_duration() steps up a unit rather than reporting a full unit's worth of the smaller"""
self.assertEqual(common.format_duration(datetime.timedelta(seconds=59.9)), '59.9 s')
self.assertEqual(common.format_duration(datetime.timedelta(seconds=59.99)), '1.0 m')
self.assertEqual(common.format_duration(datetime.timedelta(seconds=60)), '1.0 m')
self.assertEqual(common.format_duration(datetime.timedelta(minutes=59.99)), '1.0 h')
self.assertEqual(common.format_duration(datetime.timedelta(hours=23.99)), '1.0 d')

def test_format_duration_spans_seconds_to_days(self):
"""Test that format_duration() gives one decimal place and a unit symbol across its whole range"""
rendered = [common.format_duration(datetime.timedelta(seconds=seconds))
for seconds in [0, 0.05, 3.4, 59.9, 60, 3599, 3600, 86399, 86400, 12096000, 864000000]]
self.assertEqual(rendered, ['0.0 s', '0.1 s', '3.4 s', '59.9 s', '1.0 m', '1.0 h', '1.0 h',
'1.0 d', '1.0 d', '140.0 d', '10000.0 d'])
for cell in rendered:
self.assertRegex(cell, r'^\d+\.\d [smhd]$')

def test_format_duration_from_str(self):
"""Test that format_duration() accepts the str() representation of a timedelta"""
for delta in [datetime.timedelta(seconds=3.42),
datetime.timedelta(hours=13, minutes=7, seconds=6.5),
datetime.timedelta(days=1, seconds=12),
datetime.timedelta(days=2, hours=3)]:
self.assertEqual(common.format_duration(str(delta)), common.format_duration(delta))

def test_format_duration_uninterpretable(self):
"""Test that format_duration() reports an absent or negative duration as an empty string"""
for duration in [None, '', 'not a duration', 24, datetime.timedelta(seconds=-1)]:
self.assertEqual(common.format_duration(duration), '')

def test_format_duration_reads_a_string_exactly_as_timedelta_from_str_does(self):
"""Test that format_duration() and timedelta_from_str() agree on every duration string"""
for time_str in ['0:00:00', '0:00:00.500000', '0:00:03.4', '0:00:03.420000', '0:00:16.8', '0:00:18.1',
'0:17:05', '13:07:06.500000', '1 day, 0:00:00', '2 days, 3:00:00', '400 days, 0:00:01',
'-1 day, 23:59:59', '-3 days, 5:00:00', '1hr2m3s', '45s',
'', ' ', 'not a duration', '3:04', 'None']:
self.assertEqual(common.format_duration(time_str),
common.format_duration(common.timedelta_from_str(time_str)),
msg=f'{time_str!r} is read differently by the two entry points')
self.assertEqual([common.format_duration(time_str) for time_str in ['0:00:03.4', '0:00:16.8', '0:00:18.1']],
['3.4 s', '16.8 s', '18.1 s'])
self.assertEqual(common.format_duration('0:00:00'), '0.0 s')
self.assertEqual(common.format_duration('0:00:00.500000'), '0.5 s')
self.assertEqual(common.format_duration('2 days, 3:00:00'), '2.1 d')
self.assertEqual(common.format_duration('-1 day, 23:59:59'), '')
self.assertEqual(common.timedelta_from_str('-1 day, 23:59:59'), datetime.timedelta(seconds=-1))
self.assertEqual(common.format_duration('not a duration'), '')

def test_format_duration_does_not_warn_for_an_absent_duration(self):
"""Test that format_duration() reports an absent duration quietly, without a parse warning"""
with self.assertNoLogs('arc', level='WARNING'):
for duration in [None, '', ' ', datetime.timedelta(seconds=-1)]:
self.assertEqual(common.format_duration(duration), '')

def test_format_table(self):
"""Test the format_table() function"""
table = common.format_table(headers=['Label', ('H298', '(kJ/mol)')],
rows=[['CH4', '-74.60'], ['a longer label', '1.00']],
alignments='<>',
)
self.assertEqual(table, ['Label H298',
' (kJ/mol)',
'-------------- --------',
'CH4 -74.60',
'a longer label 1.00'])

def test_format_table_column_widths(self):
"""Test that format_table() sizes each column to its widest entry, header or cell"""
table = common.format_table(headers=['A', 'BBBBB'], rows=[['CCC', 'D']], separator='|', rule_char='')
self.assertEqual(table, ['A |BBBBB', 'CCC|D'])

def test_format_table_no_rows(self):
"""Test that format_table() renders the header alone when there are no rows"""
self.assertEqual(common.format_table(headers=['A', 'BB'], rows=[]), ['A BB', '- --'])

def test_format_table_raises_on_a_ragged_row(self):
"""Test that format_table() rejects a row or an alignment string that does not match the headers"""
with self.assertRaises(InputError):
common.format_table(headers=['A', 'B'], rows=[['1']])
with self.assertRaises(InputError):
common.format_table(headers=['A', 'B'], rows=[['1', '2']], alignments='<')

def test_format_table_raises_input_error_on_a_bad_cell_or_alignment(self):
"""Test that format_table() reports a non-string cell or an unknown alignment as an InputError"""
for row in [[None, '2'], [1, '2'], [['1'], '2']]:
with self.assertRaises(InputError):
common.format_table(headers=['A', 'B'], rows=[row])
with self.assertRaises(InputError):
common.format_table(headers=['A', 'B'], rows=[['1', '2']], alignments='<x')

def test_format_table_with_an_empty_header(self):
"""Test that format_table() renders a column with an empty multi-line header and no rows"""
self.assertEqual(common.format_table(headers=[()], rows=[]), [''])

def test_check_ess_settings(self):
"""Test the check_ess_settings function"""
server_names = list(servers.keys())
Expand Down Expand Up @@ -1289,6 +1410,41 @@ def test_timedelta_from_str(self):
self.assertIn('0:00:00.5', str_delta)
reconstructed_delta = common.timedelta_from_str(str_delta)
self.assertIsInstance(reconstructed_delta, datetime.timedelta)
self.assertEqual(reconstructed_delta, delta)

self.assertEqual(common.timedelta_from_str('0:00:03.420000'),
datetime.timedelta(seconds=3, microseconds=420000))
self.assertEqual(common.timedelta_from_str('2 days, 3:00:00'), datetime.timedelta(days=2, hours=3))
self.assertEqual(common.timedelta_from_str('1 day, 0:00:00'), datetime.timedelta(days=1))
self.assertEqual(common.timedelta_from_str('400 days, 0:00:01'), datetime.timedelta(days=400, seconds=1))
self.assertEqual(common.timedelta_from_str('-1 day, 23:59:59'), datetime.timedelta(seconds=-1))

self.assertEqual(common.timedelta_from_str('0:00:00'), datetime.timedelta(0))
self.assertIsNotNone(common.timedelta_from_str('0:00:00'))

self.assertEqual(common.timedelta_from_str('1hr2m3s'), datetime.timedelta(hours=1, minutes=2, seconds=3))
self.assertEqual(common.timedelta_from_str('45s'), datetime.timedelta(seconds=45))
self.assertEqual(common.timedelta_from_str('2m'), datetime.timedelta(minutes=2))

for time_str in ['', ' ', '0', 'None', 'nonsense', '1:2', '3:04', 'hrms', '0:00:03.42x']:
self.assertIsNone(common.timedelta_from_str(time_str), msg=f'{time_str!r} must not parse')
self.assertIsNone(common.timedelta_from_str(None))
self.assertIsNone(common.timedelta_from_str(5))

def test_timedelta_from_str_round_trip(self):
"""Test that timedelta_from_str() inverts str() for a spread of durations"""
for delta in [datetime.timedelta(0),
datetime.timedelta(microseconds=1),
datetime.timedelta(seconds=0.5),
datetime.timedelta(seconds=3, microseconds=420000),
datetime.timedelta(minutes=17, seconds=5),
datetime.timedelta(hours=26, minutes=3, seconds=9, microseconds=123456),
datetime.timedelta(days=2, hours=3),
datetime.timedelta(days=400, seconds=1),
datetime.timedelta(seconds=-1),
datetime.timedelta(days=-3, hours=5),
]:
self.assertEqual(common.timedelta_from_str(str(delta)), delta, msg=f'failed to round-trip {str(delta)!r}')

def test_torsions_to_scans(self):
"""Test the torsions_to_scans() function"""
Expand Down
Loading
Loading