lint: Second pass with ruff --fix --unsafe-fixes

Author: JDBetteridge

benchmarks/user/advisor/advisor_logging.py

@@ -9,18 +9,18 @@ def check(cond, msg):
 
 
 def err(msg):
-    print('\033[1;37;31m%s\033[0m' % msg)  # print in RED
+    print(f'\033[1;37;31m{msg}\033[0m')  # print in RED
 
 
 def log(msg):
-    print('\033[1;37;32m%s\033[0m' % msg)  # print in GREEN
+    print(f'\033[1;37;32m{msg}\033[0m')  # print in GREEN
 
 
 @contextmanager
 def progress(msg):
-    print('\033[1;37;32m%s ... \033[0m' % msg, end='', flush=True)  # print in GREEN
+    print(f'\033[1;37;32m{msg} ... \033[0m', end='', flush=True)  # print in GREEN
     yield
-    print('\033[1;37;32m%s\033[0m' % 'Done!')
+    print('\033[1;37;32m{}\033[0m'.format('Done!'))
 
 
 def log_process(process, logger):

benchmarks/user/advisor/roofline.py

@@ -205,7 +205,7 @@ def roofline(name, project, scale, precision, mode, th):
     log(f'\nFigure saved in {figpath}{name}.pdf.')
 
     # Save the JSON file
-    with open('%s.json' % name, 'w') as f:
+    with open(f'{name}.json', 'w') as f:
         f.write(json.dumps(roofline_data))
 
     log(f'\nJSON file saved as {name}.json.')

benchmarks/user/advisor/run_advisor.py

@@ -132,7 +132,7 @@ def run_with_advisor(path, output, name, exec_args):
     # Before collecting the `survey` and `tripcounts` a "pure" python run
     # to warmup the jit cache is preceded
 
-    log('Starting Intel Advisor\'s `roofline` analysis for `%s`' % name)
+    log(f'Starting Intel Advisor\'s `roofline` analysis for `{name}`')
     dt = datetime.datetime.now()
 
     # Set up a file logger that will track the output of the advisor profiling
@@ -142,7 +142,7 @@ def run_with_advisor(path, output, name, exec_args):
     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)
-    advixe_handler = logging.FileHandler('%s/%s_%s.log' % (output, name, logger_datetime))
+    advixe_handler = logging.FileHandler(f'{output}/{name}_{logger_datetime}.log')
     advixe_handler.setFormatter(advixe_formatter)
     advixe_logger.addHandler(advixe_handler)
 

benchmarks/user/benchmark.py

@@ -74,7 +74,7 @@ def run_op(solver, operator, **options):
     try:
         op = getattr(solver, operator)
     except AttributeError:
-        raise AttributeError("Operator %s not implemented for %s" % (operator, solver))
+        raise AttributeError(f"Operator {operator} not implemented for {solver}")
 
     # This is a bit ugly but not sure how to make clean input creation for different op
     if operator == "forward":
@@ -95,7 +95,7 @@ def run_op(solver, operator, **options):
         args = args[:-1]
         return op(*args, **options)
     else:
-        raise ValueError("Unrecognized operator %s" % operator)
+        raise ValueError(f"Unrecognized operator {operator}")
 
 
 @click.group()
@@ -157,15 +157,14 @@ def from_opt(ctx, param, value):
             # E.g., `('advanced', {'par-tile': True})`
             value = eval(value)
             if not isinstance(value, tuple) and len(value) >= 1:
-                raise click.BadParameter("Invalid choice `%s` (`opt` must be "
-                                         "either str or tuple)" % str(value))
+                raise click.BadParameter(f"Invalid choice `{str(value)}` (`opt` must be "
+                                         "either str or tuple)")
             opt = value[0]
         except NameError:
             # E.g. `'advanced'`
             opt = value
         if opt not in configuration._accepted['opt']:
-            raise click.BadParameter("Invalid choice `%s` (choose from %s)"
-                                     % (opt, str(configuration._accepted['opt'])))
+            raise click.BadParameter("Invalid choice `{}` (choose from {})".format(opt, str(configuration._accepted['opt'])))
         return value
 
     def config_blockshape(ctx, param, value):
@@ -181,18 +180,18 @@ 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 _i, block_shape in enumerate(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()]
                 levels = [bs[x:x+3] for x in range(0, len(bs), 3)]
                 if any(len(level) != 3 for level in levels):
                     raise ValueError("Expected 3 entries per block shape level, but got "
-                                     "one level with less than 3 entries (`%s`)" % levels)
+                                     f"one level with less than 3 entries (`{levels}`)")
                 normalized_value.append(levels)
             if not all_equal(len(i) for i in normalized_value):
                 raise ValueError("Found different block shapes with incompatible "
-                                 "number of levels (`%s`)" % normalized_value)
+                                 f"number of levels (`{normalized_value}`)")
             configuration['opt-options']['blocklevels'] = len(normalized_value[0])
         else:
             normalized_value = []
@@ -205,8 +204,7 @@ 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 `%s`)"
-                        % str(ctx.params['block_shape']))
+                warning("Skipping autotuning (using explicit block-shape `{}`)".format(str(ctx.params['block_shape'])))
                 level = False
             else:
                 # Make sure to always run in preemptive mode
@@ -305,11 +303,11 @@ def run(problem, **kwargs):
 
     dumpfile = kwargs.pop('dump_norms')
     if dumpfile:
-        norms = ["'%s': %f" % (i.name, norm(i)) for i in retval[:-1]
+        norms = [f"'{i.name}': {norm(i):f}" for i in retval[:-1]
                  if isinstance(i, DiscreteFunction)]
         if rank == 0:
             with open(dumpfile, 'w') as f:
-                f.write("{%s}" % ', '.join(norms))
+                f.write("{{{}}}".format(', '.join(norms)))
 
     return retval
 
@@ -343,13 +341,13 @@ def run_jit_backdoor(problem, **kwargs):
     op = solver.op_fwd()
 
     # Get the filename in the JIT cache
-    cfile = "%s.c" % str(op._compiler.get_jit_dir().joinpath(op._soname))
+    cfile = f"{str(op._compiler.get_jit_dir().joinpath(op._soname))}.c"
 
     if not os.path.exists(cfile):
         # First time we run this problem, let's generate and jit-compile code
         op.cfunction
-        info("You may now edit the generated code in `%s`. "
-             "Then save the file, and re-run this benchmark." % cfile)
+        info(f"You may now edit the generated code in `{cfile}`. "
+             "Then save the file, and re-run this benchmark.")
         return
 
     info("Running wave propagation Operator...")
@@ -364,7 +362,7 @@ def _run_jit_backdoor():
     if dumpnorms:
         for i in retval[:-1]:
             if isinstance(i, DiscreteFunction):
-                info("'%s': %f" % (i.name, norm(i)))
+                info(f"'{i.name}': {norm(i):f}")
 
     return retval
 

conftest.py

@@ -17,6 +17,7 @@
     FindNodes, FindSymbols, Iteration, ParallelBlock, retrieve_iteration_tree
 )
 from devito.tools import as_tuple
+import contextlib
 
 try:
     from mpi4py import MPI  # noqa
@@ -40,29 +41,28 @@ def skipif(items, whole_module=False):
     accepted.update({'nodevice', 'noomp'})
     unknown = sorted(set(items) - accepted)
     if unknown:
-        raise ValueError("Illegal skipif argument(s) `%s`" % unknown)
+        raise ValueError(f"Illegal skipif argument(s) `{unknown}`")
     skipit = False
     for i in items:
         # Skip if won't run on GPUs
         if i == 'device' and isinstance(configuration['platform'], Device):
-            skipit = "device `%s` unsupported" % configuration['platform'].name
+            skipit = "device `{}` unsupported".format(configuration['platform'].name)
             break
         # Skip if won't run on a specific GPU backend
         langs = configuration._accepted['language']
-        if any(i == 'device-%s' % l and configuration['language'] == l for l in langs)\
+        if any(i == f'device-{l}' and configuration['language'] == l for l in langs)\
                 and isinstance(configuration['platform'], Device):
-            skipit = "language `%s` for device unsupported" % configuration['language']
+            skipit = "language `{}` for device unsupported".format(configuration['language'])
             break
-        if any(i == 'device-%s' % k and isinstance(configuration['compiler'], v)
+        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 `%s` for device unsupported" % configuration['compiler']
+            skipit = "compiler `{}` for device unsupported".format(configuration['compiler'])
             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 `%s`" %
-                      configuration['platform'].name)
+            skipit = ("must run on device, but currently on `{}`".format(configuration['platform'].name))
             break
         # Skip if it won't run with nvc on CPU backend
         if i == 'cpu64-nvc' and \
@@ -137,9 +137,9 @@ def EVAL(exprs, *args):
 
 def get_testname(item):
     if item.cls is not None:
-        return "%s::%s::%s" % (item.fspath, item.cls.__name__, item.name)
+        return f"{item.fspath}::{item.cls.__name__}::{item.name}"
     else:
-        return "%s::%s" % (item.fspath, item.name)
+        return f"{item.fspath}::{item.name}"
 
 
 def set_run_reset(env_vars, call):
@@ -179,7 +179,7 @@ def parallel(item, m):
         if len(m) == 2:
             nprocs, scheme = m
         else:
-            raise ValueError("Can't run test: unexpected mode `%s`" % m)
+            raise ValueError(f"Can't run test: unexpected mode `{m}`")
 
     env_vars = {'DEVITO_MPI': scheme}
 
@@ -247,10 +247,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 contextlib.suppress(ValueError):
         inside_pytest_marker = int(inside_pytest_marker)
-    except ValueError:
-        pass
 
     if inside_pytest_marker:
         outcome = yield
@@ -281,10 +279,8 @@ def pytest_runtest_makereport(item, call):
     result = outcome.get_result()
 
     inside_pytest_marker = os.environ.get('DEVITO_PYTEST_FLAG', 0)
-    try:
+    with contextlib.suppress(ValueError):
         inside_pytest_marker = int(inside_pytest_marker)
-    except ValueError:
-        pass
     if inside_pytest_marker:
         return
 

devito/arch/archinfo.py

@@ -15,31 +15,79 @@
 
 from devito.logger import warning
 from devito.tools import all_equal, as_tuple, memoized_func
-
-__all__ = ['platform_registry', 'get_cpu_info', 'get_gpu_info', 'get_visible_devices',
-           'get_nvidia_cc', 'get_cuda_path', 'get_hip_path', 'check_cuda_runtime',
-           'get_m1_llvm_path', 'get_advisor_path', 'Platform', 'Cpu64', 'Intel64',
-           'IntelSkylake', 'Amd', 'Arm', 'Power', 'Device', 'NvidiaDevice',
-           'AmdDevice', 'IntelDevice',
-           # Brand-agnostic
-           'ANYCPU', 'ANYGPU',
-           # Intel CPUs
-           'INTEL64', 'SNB', 'IVB', 'HSW', 'BDW', 'KNL', 'KNL7210',
-           'SKX', 'KLX', 'CLX', 'CLK', 'SPR',
-           # AMD CPUs
-           'AMD',
-           # ARM CPUs
-           'ARM', 'AppleArm', 'M1', 'M2', 'M3',
-           'Graviton', 'GRAVITON2', 'GRAVITON3', 'GRAVITON4',
-           'Cortex', 'NvidiaArm', 'GRACE',
-           # Other legacy CPUs
-           'POWER8', 'POWER9',
-           # Generic GPUs
-           'AMDGPUX', 'NVIDIAX', 'INTELGPUX',
-           # Nvidia GPUs
-           'VOLTA', 'AMPERE', 'HOPPER', 'BLACKWELL',
-           # Intel GPUs
-           'PVC', 'INTELGPUMAX', 'MAX1100', 'MAX1550']
+import contextlib
+
+__all__ = [
+    # AMD CPUs
+    'AMD',
+    # Generic GPUs
+    'AMDGPUX',
+    'AMPERE',
+    # Brand-agnostic
+    'ANYCPU',
+    'ANYGPU',
+    # ARM CPUs
+    'ARM',
+    'BDW',
+    'BLACKWELL',
+    'CLK',
+    'CLX',
+    'GRACE',
+    'GRAVITON2',
+    'GRAVITON3',
+    'GRAVITON4',
+    'HOPPER',
+    'HSW',
+    # Intel CPUs
+    'INTEL64',
+    'INTELGPUMAX',
+    'INTELGPUX',
+    'IVB',
+    'KLX',
+    'KNL',
+    'KNL7210',
+    'M1',
+    'M2',
+    'M3',
+    'MAX1100',
+    'MAX1550',
+    'NVIDIAX',
+    # Other legacy CPUs
+    'POWER8',
+    'POWER9',
+    # Intel GPUs
+    'PVC',
+    'SKX',
+    'SNB',
+    'SPR',
+    # Nvidia GPUs
+    'VOLTA',
+    'Amd',
+    'AmdDevice',
+    'AppleArm',
+    'Arm',
+    'Cortex',
+    'Cpu64',
+    'Device',
+    'Graviton',
+    'Intel64',
+    'IntelDevice',
+    'IntelSkylake',
+    'NvidiaArm',
+    'NvidiaDevice',
+    'Platform',
+    'Power',
+    'check_cuda_runtime',
+    'get_advisor_path',
+    'get_cpu_info',
+    'get_cuda_path',
+    'get_gpu_info',
+    'get_hip_path',
+    'get_m1_llvm_path',
+    'get_nvidia_cc',
+    'get_visible_devices',
+    'platform_registry',
+]
 
 
 @memoized_func
@@ -387,7 +435,7 @@ def cbk(deviceid=0):
                     return None
                 return cbk
 
-            gpu_info['mem.%s' % i] = make_cbk(i)
+            gpu_info[f'mem.{i}'] = make_cbk(i)
 
         gpu_info['architecture'] = 'unspecified'
         gpu_info['vendor'] = 'INTEL'
@@ -745,7 +793,7 @@ def __str__(self):
         return self.name
 
     def __repr__(self):
-        return "TargetPlatform[%s]" % self.name
+        return f"TargetPlatform[{self.name}]"
 
     def _detect_isa(self):
         return 'unknown'
@@ -1139,10 +1187,8 @@ def march(cls):
         try:
             p1 = Popen(['offload-arch'], stdout=PIPE, stderr=PIPE)
         except OSError:
-            try:
+            with contextlib.suppress(OSError):
                 p1 = Popen(['mygpu', '-d', fallback], stdout=PIPE, stderr=PIPE)
-            except OSError:
-                pass
             return fallback
 
         output, _ = p1.communicate()
@@ -1185,7 +1231,7 @@ def node_max_mem_trans_nbytes(platform):
     elif isinstance(platform, Device):
         return max(Cpu64.max_mem_trans_nbytes, mmtb0)
     else:
-        assert False, f"Unknown platform type: {type(platform)}"
+        raise AssertionError(f"Unknown platform type: {type(platform)}")
 
 
 # CPUs