misc: Lint benchmarks

Author: JDBetteridge

benchmarks/regression/benchmarks/arguments.py

@@ -10,9 +10,9 @@ class Processing:
     def setup(self):
         grid = Grid(shape=(5, 5, 5))
 
-        funcs = [Function(name='f%d' % n, grid=grid) for n in range(30)]
-        tfuncs = [TimeFunction(name='u%d' % n, grid=grid) for n in range(30)]
-        stfuncs = [SparseTimeFunction(name='su%d' % n, grid=grid, npoint=1, nt=100)
+        funcs = [Function(name=f'f{n}', grid=grid) for n in range(30)]
+        tfuncs = [TimeFunction(name=f'u{n}', grid=grid) for n in range(30)]
+        stfuncs = [SparseTimeFunction(name=f'su{n}', grid=grid, npoint=1, nt=100)
                    for n in range(30)]
         v = TimeFunction(name='v', grid=grid, space_order=2)
 

benchmarks/user/advisor/run_advisor.py

@@ -140,8 +140,7 @@ def run_with_advisor(path, output, name, exec_args):
     advixe_logger.setLevel(logging.INFO)
 
     advixe_formatter = logging.Formatter('%(asctime)s: %(message)s')
-    logger_datetime = '%d.%d.%d.%d.%d.%d' % (dt.year, dt.month,
-                                             dt.day, dt.hour, dt.minute, dt.second)
+    logger_datetime = f'{dt.year}.{dt.month}.{dt.day}.{dt.hour}.{dt.minute}.{dt.second}'
     advixe_handler = logging.FileHandler(f'{output}/{name}_{logger_datetime}.log')
     advixe_handler.setFormatter(advixe_formatter)
     advixe_logger.addHandler(advixe_handler)

benchmarks/user/benchmark.py

@@ -1,4 +1,5 @@
 import os
+from contextlib import suppress
 
 import click
 import numpy as np
@@ -73,8 +74,10 @@ def run_op(solver, operator, **options):
     # Get the operator if exist
     try:
         op = getattr(solver, operator)
-    except AttributeError:
-        raise AttributeError(f"Operator {operator} not implemented for {solver}")
+    except AttributeError as e:
+        raise AttributeError(
+            f"Operator {operator} not implemented for {solver}"
+        ) from e
 
     # This is a bit ugly but not sure how to make clean input creation for different op
     if operator == "forward":
@@ -164,7 +167,10 @@ def from_opt(ctx, param, value):
             # E.g. `'advanced'`
             opt = value
         if opt not in configuration._accepted['opt']:
-            raise click.BadParameter("Invalid choice `{}` (choose from {})".format(opt, str(configuration._accepted['opt'])))
+            raise click.BadParameter(
+                f'Invalid choice `{opt} '
+                f'(choose from {str(configuration._accepted["opt"])})'
+            )
         return value
 
     def config_blockshape(ctx, param, value):
@@ -180,7 +186,7 @@ def config_blockshape(ctx, param, value):
             # 1. integers, not strings
             # 2. sanity check the (hierarchical) blocking shape
             normalized_value = []
-            for i, block_shape in enumerate(value):
+            for block_shape in value:
                 # If hierarchical blocking is activated, say with N levels, here in
                 # `bs` we expect to see 3*N entries
                 bs = [int(x) for x in block_shape.split()]
@@ -204,7 +210,10 @@ def config_autotuning(ctx, param, value):
         elif value != 'off':
             # Sneak-peek at the `block-shape` -- if provided, keep auto-tuning off
             if ctx.params['block_shape']:
-                warning("Skipping autotuning (using explicit block-shape `{}`)".format(str(ctx.params['block_shape'])))
+                warning(
+                    'Skipping autotuning'
+                    f'(using explicit block-shape `{str(ctx.params["block_shape"])}`)'
+                )
                 level = False
             else:
                 # Make sure to always run in preemptive mode
@@ -272,8 +281,8 @@ def run(problem, **kwargs):
     # Note: the following piece of code is horribly *hacky*, but it works for now
     for i, block_shape in enumerate(block_shapes):
         for n, level in enumerate(block_shape):
-            for d, s in zip(['x', 'y', 'z'], level):
-                options['%s%d_blk%d_size' % (d, i, n)] = s
+            for d, s in zip(['x', 'y', 'z'], level, strict=True):
+                options[f'{d}{i}_blk{n}_size'] = s
 
     solver = setup(space_order=space_order, time_order=time_order, **kwargs)
     if warmup:
@@ -345,7 +354,7 @@ def run_jit_backdoor(problem, **kwargs):
 
     if not os.path.exists(cfile):
         # First time we run this problem, let's generate and jit-compile code
-        op.cfunction
+        _ = op.cfunction
         info(f"You may now edit the generated code in `{cfile}`. "
              "Then save the file, and re-run this benchmark.")
         return
@@ -403,9 +412,10 @@ def test(problem, **kwargs):
         set_log_level('DEBUG', comm=MPI.COMM_WORLD)
 
         if MPI.COMM_WORLD.size > 1 and not configuration['mpi']:
-            warning("It seems that you're running over MPI with %d processes, but "
-                    "DEVITO_MPI is unset. Setting `DEVITO_MPI=basic`..."
-                    % MPI.COMM_WORLD.size)
+            warning(
+                f'It seems that you are running over MPI with {MPI.COMM_WORLD.size} '
+                'processes, but DEVITO_MPI is unset. Setting `DEVITO_MPI=basic`...'
+            )
             configuration['mpi'] = 'basic'
     except (TypeError, ModuleNotFoundError):
         # MPI not available
@@ -417,8 +427,6 @@ def test(problem, **kwargs):
 
     benchmark(standalone_mode=False)
 
-    try:
+    # In case MPI not available
+    with suppress(TypeError):
         MPI.Finalize()
-    except TypeError:
-        # MPI not available
-        pass

conftest.py

@@ -1,5 +1,6 @@
 import os
 import sys
+from contextlib import suppress
 from subprocess import check_call
 
 import pytest
@@ -51,17 +52,18 @@ def skipif(items, whole_module=False):
         langs = configuration._accepted['language']
         if any(i == f'device-{l}' and configuration['language'] == l for l in langs)\
                 and isinstance(configuration['platform'], Device):
-            skipit = "language `{}` for device unsupported".format(configuration['language'])
+            skipit = f'language `{configuration["language"]}` for device unsupported'
             break
         if any(i == f'device-{k}' and isinstance(configuration['compiler'], v)
                for k, v in compiler_registry.items()) and\
                 isinstance(configuration['platform'], Device):
-            skipit = "compiler `{}` for device unsupported".format(configuration['compiler'])
+            skipit = f'compiler `{configuration["compiler"]}` for device unsupported'
             break
         # Skip if must run on GPUs but not currently on a GPU
         if i in ('nodevice', 'nodevice-omp', 'nodevice-acc') and\
                 not isinstance(configuration['platform'], Device):
-            skipit = ("must run on device, but currently on `{}`".format(configuration['platform'].name))
+            skipit = 'must run on device, but currently on '
+            skipit += f'`{configuration["platform"].name}`'
             break
         # Skip if it won't run with nvc on CPU backend
         if i == 'cpu64-nvc' and \
@@ -193,7 +195,7 @@ def parallel(item, m):
     ]
     if nprocs > 1:
         args.extend([
-            ":", "-n", "%d" % (nprocs - 1), pyversion, "-m", "pytest",
+            ":", "-n", str(nprocs - 1), pyversion, "-m", "pytest",
             "-s", "--runxfail", "-v", "--timeout=600", "--timeout-method=thread",
             "-o faulthandler_timeout=660", testname
         ])
@@ -253,10 +255,8 @@ def pytest_generate_tests(metafunc):
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_runtest_call(item):
     inside_pytest_marker = os.environ.get('DEVITO_PYTEST_FLAG', 0)
-    try:
+    with suppress(ValueError):
         inside_pytest_marker = int(inside_pytest_marker)
-    except ValueError:
-        pass
 
     if inside_pytest_marker:
         outcome = yield
@@ -287,15 +287,13 @@ def pytest_runtest_makereport(item, call):
     result = outcome.get_result()
 
     inside_pytest_marker = os.environ.get('DEVITO_PYTEST_FLAG', 0)
-    try:
+    with suppress(ValueError):
         inside_pytest_marker = int(inside_pytest_marker)
-    except ValueError:
-        pass
     if inside_pytest_marker:
         return
 
     if item.get_closest_marker("parallel") or \
-       item.get_closest_marker("decoupler"):
+       item.get_closest_marker("decoupler"):  # noqa: SIM102
         if call.when == 'call' and result.outcome == 'skipped':
             result.outcome = 'passed'