STY: Apply ruff/flynt rule FLY002

FLY002 Consider f-string instead of string join

nipype/algorithms/rapidart.py

@@ -397,19 +397,13 @@ def _get_output_filenames(self, motionfile, output_dir):
         else:
             raise Exception("Unknown type of file")
         _, filename, ext = split_filename(infile)
-        artifactfile = os.path.join(
-            output_dir, "".join(("art.", filename, "_outliers.txt"))
-        )
-        intensityfile = os.path.join(
-            output_dir, "".join(("global_intensity.", filename, ".txt"))
-        )
-        statsfile = os.path.join(output_dir, "".join(("stats.", filename, ".txt")))
-        normfile = os.path.join(output_dir, "".join(("norm.", filename, ".txt")))
-        plotfile = os.path.join(
-            output_dir, "".join(("plot.", filename, ".", self.inputs.plot_type))
-        )
-        displacementfile = os.path.join(output_dir, "".join(("disp.", filename, ext)))
-        maskfile = os.path.join(output_dir, "".join(("mask.", filename, ext)))
+        artifactfile = os.path.join(output_dir, f"art.{filename}_outliers.txt")
+        intensityfile = os.path.join(output_dir, f"global_intensity.{filename}.txt")
+        statsfile = os.path.join(output_dir, f"stats.{filename}.txt")
+        normfile = os.path.join(output_dir, f"norm.{filename}.txt")
+        plotfile = os.path.join(output_dir, f"plot.{filename}.{self.inputs.plot_type}")
+        displacementfile = os.path.join(output_dir, f"disp.{filename}{ext}")
+        maskfile = os.path.join(output_dir, f"mask.{filename}{ext}")
         return (
             artifactfile,
             intensityfile,
@@ -760,7 +754,7 @@ def _get_output_filenames(self, motionfile, output_dir):
         """
         (_, filename) = os.path.split(motionfile)
         (filename, _) = os.path.splitext(filename)
-        corrfile = os.path.join(output_dir, "".join(("qa.", filename, "_stimcorr.txt")))
+        corrfile = os.path.join(output_dir, f"qa.{filename}_stimcorr.txt")
         return corrfile
 
     def _stimcorr_core(self, motionfile, intensityfile, designmatrix, cwd=None):

nipype/interfaces/afni/base.py

@@ -292,7 +292,7 @@ def _gen_fname(self, basename, cwd=None, suffix=None, change_ext=True, ext=None)
         if ext is None:
             ext = Info.output_type_to_ext(self.inputs.outputtype)
         if change_ext:
-            suffix = "".join((suffix, ext)) if suffix else ext
+            suffix = f"{suffix}{ext}" if suffix else ext
 
         if suffix is None:
             suffix = ""

nipype/interfaces/ants/resampling.py

@@ -134,7 +134,7 @@ def _list_outputs(self):
         outputs = self._outputs().get()
         _, name, ext = split_filename(os.path.abspath(self.inputs.input_image))
         outputs["output_image"] = os.path.join(
-            os.getcwd(), "".join((name, self.inputs.out_postfix, ext))
+            os.getcwd(), f"{name}{self.inputs.out_postfix}{ext}"
         )
         return outputs
 
@@ -254,7 +254,7 @@ class WarpImageMultiTransform(ANTSCommand):
     def _gen_filename(self, name):
         if name == "output_image":
             _, name, ext = split_filename(os.path.abspath(self.inputs.input_image))
-            return "".join((name, self.inputs.out_postfix, ext))
+            return f"{name}{self.inputs.out_postfix}{ext}"
         return None
 
     def _format_arg(self, opt, spec, val):

nipype/interfaces/base/core.py

@@ -709,7 +709,7 @@ def version_from_command(self, flag="-v", cmd=None):
             out_environ = self._get_environ()
             env.update(out_environ)
             proc = sp.Popen(
-                " ".join((cmd, flag)),
+                f"{cmd} {flag}",
                 shell=True,
                 env=canonicalize_env(env),
                 stdout=sp.PIPE,

nipype/interfaces/base/specs.py

@@ -136,7 +136,7 @@ def _deprecated_warn(self, obj, name, old, new):
                 msg3 = "It has been replaced by %s." % trait_spec.new_name
             else:
                 msg3 = ""
-            msg = " ".join((msg1, msg2, msg3))
+            msg = f"{msg1} {msg2} {msg3}"
             if Version(str(trait_spec.deprecated)) < self.package_version:
                 raise TraitError(msg)
             else:

nipype/interfaces/brainsuite/brainsuite.py

@@ -1801,7 +1801,7 @@ def getFileName(inputName, suffix):
     dotRegex = regex.compile("[^.]+")
     # extract between last slash and first period
     inputNoExtension = dotRegex.findall(fullInput)[0]
-    return os.path.abspath("".join((inputNoExtension, suffix)))
+    return os.path.abspath(f"{inputNoExtension}{suffix}")
 
 
 def l_outputs(self):

nipype/interfaces/cat12/surface.py

@@ -145,8 +145,8 @@ def _list_outputs(self):
                             outputs[name_hemisphere] = []
                         if not isdefined(outputs[all_files_hemisphere]):
                             outputs[all_files_hemisphere] = []
-                        generated_filename = ".".join(
-                            [hemisphere, parameter_name, original_filename]
+                        generated_filename = (
+                            f"{hemisphere}.{parameter_name}.{original_filename}"
                         )
                         outputs[name_hemisphere].append(
                             os.path.join(pth, generated_filename)

nipype/interfaces/dtitk/base.py

@@ -89,7 +89,7 @@ def _gen_fname(self, basename, cwd=None, suffix=None, change_ext=True, ext=None)
             ext = Info.output_type_to_ext(self.inputs.output_type)
         if change_ext:
             if suffix:
-                suffix = "".join((suffix, ext))
+                suffix = f"{suffix}{ext}"
             else:
                 suffix = ext
         if suffix is None:

nipype/interfaces/freesurfer/model.py

@@ -673,7 +673,7 @@ def _list_outputs(self):
                 outfile = fname_presuffix(
                     self.inputs.in_file,
                     newpath=os.getcwd(),
-                    suffix=".".join(("_thresh", self.inputs.out_type)),
+                    suffix=f"_thresh.{self.inputs.out_type}",
                     use_ext=False,
                 )
             else:

nipype/interfaces/freesurfer/preprocess.py

@@ -2540,7 +2540,7 @@ def _format_arg(self, name, spec, value):
                     cmd = " ".join((cmd, "-fa %.1f" % self.inputs.flip_list[i]))
                 if isdefined(self.inputs.xfm_list):
                     cmd = " ".join((cmd, "-at %s" % self.inputs.xfm_list[i]))
-                cmd = " ".join((cmd, file))
+                cmd = f"{cmd} {file}"
             return cmd
         return super()._format_arg(name, spec, value)
 

nipype/interfaces/fsl/base.py

@@ -237,7 +237,7 @@ def _gen_fname(self, basename, cwd=None, suffix=None, change_ext=True, ext=None)
             ext = Info.output_type_to_ext(self.inputs.output_type)
         if change_ext:
             if suffix:
-                suffix = "".join((suffix, ext))
+                suffix = f"{suffix}{ext}"
             else:
                 suffix = ext
         if suffix is None:

nipype/interfaces/fsl/tests/test_preprocess.py

@@ -306,21 +306,21 @@ def test_flirt(setup_flirt):
         # Handle autogeneration of outfile
         pth, fname, ext = split_filename(infile)
         outfile = fsl_name(fsl.FLIRT(), "%s_flirt" % fname)
-        outfile = " ".join(["-out", outfile])
+        outfile = f"-out {outfile}"
         # Handle autogeneration of outmatrix
         outmatrix = "%s_flirt.mat" % fname
-        outmatrix = " ".join(["-omat", outmatrix])
+        outmatrix = f"-omat {outmatrix}"
         # Build command line
-        cmdline = " ".join([cmdline, outfile, outmatrix, param])
+        cmdline = f"{cmdline} {outfile} {outmatrix} {param}"
         flirter = fsl.FLIRT(in_file=infile, reference=reffile)
         setattr(flirter.inputs, key, value)
         assert flirter.cmdline == cmdline
 
     # Test OutputSpec
     flirter = fsl.FLIRT(in_file=infile, reference=reffile)
     pth, fname, ext = split_filename(infile)
-    flirter.inputs.out_file = "".join(["foo", ext])
-    flirter.inputs.out_matrix_file = "".join(["bar", ext])
+    flirter.inputs.out_file = f"foo{ext}"
+    flirter.inputs.out_matrix_file = f"bar{ext}"
     outs = flirter._list_outputs()
     assert outs["out_file"] == os.path.join(os.getcwd(), flirter.inputs.out_file)
     assert outs["out_matrix_file"] == os.path.join(

nipype/interfaces/io.py

@@ -1820,7 +1820,7 @@ def _get_files(self, path, key, dirval, altkey=None):
             else:
                 globprefix = "*"
         keys = ensure_list(altkey) if altkey else [key]
-        globfmt = os.path.join(path, dirval, "".join((globprefix, "{}", globsuffix)))
+        globfmt = os.path.join(path, dirval, f"{globprefix}{{}}{globsuffix}")
         return [
             os.path.abspath(f) for key in keys for f in glob.glob(globfmt.format(key))
         ]

nipype/interfaces/mne/base.py

@@ -112,7 +112,7 @@ def _get_files(self, path, key, dirval, altkey=None):
         keydir = op.join(path, dirval)
         if altkey:
             key = altkey
-        globpattern = op.join(keydir, "".join((globprefix, key, globsuffix)))
+        globpattern = op.join(keydir, f"{globprefix}{key}{globsuffix}")
         return glob.glob(globpattern)
 
     def _list_outputs(self):

nipype/interfaces/niftyfit/base.py

@@ -45,5 +45,5 @@ def _gen_fname(self, basename, out_dir=None, suffix=None, ext=None):
         if ext is not None:
             final_ext = ext
         if suffix is not None:
-            final_bn = "".join((final_bn, suffix))
+            final_bn = f"{final_bn}{suffix}"
         return os.path.abspath(os.path.join(out_dir, final_bn + final_ext))

nipype/interfaces/niftyreg/base.py

@@ -132,5 +132,5 @@ def _gen_fname(self, basename, out_dir=None, suffix=None, ext=None):
         if ext is not None:
             final_ext = ext
         if suffix is not None:
-            final_bn = "".join((final_bn, suffix))
+            final_bn = f"{final_bn}{suffix}"
         return os.path.abspath(os.path.join(out_dir, final_bn + final_ext))

nipype/interfaces/petpvc.py

@@ -225,7 +225,7 @@ def _gen_fname(
             cwd = os.getcwd()
         if change_ext:
             if suffix:
-                suffix = "".join((suffix, ext))
+                suffix = f"{suffix}{ext}"
             else:
                 suffix = ext
         if suffix is None:

nipype/interfaces/spm/preprocess.py

@@ -1262,7 +1262,7 @@ def _list_outputs(self):
             outputs["normalized_source"] = self.inputs.source
         elif "write" in self.inputs.jobtype:
             if isdefined(self.inputs.write_preserve) and self.inputs.write_preserve:
-                prefixNorm = "".join(["m", self.inputs.out_prefix])
+                prefixNorm = f"m{self.inputs.out_prefix}"
             else:
                 prefixNorm = self.inputs.out_prefix
             outputs["normalized_files"] = []

nipype/pipeline/engine/utils.py

@@ -457,9 +457,9 @@ def get_print_name(node, simple_form=True):
         if len(pkglist) > 2:
             destclass = ".%s" % pkglist[2]
         if simple_form:
-            name = node.fullname + destclass
+            name = f"{node.fullname}{destclass}"
         else:
-            name = ".".join([node.fullname, interface]) + destclass
+            name = f"{node.fullname}.{interface}{destclass}"
     if simple_form:
         parts = name.split(".")
         if len(parts) > 2:
@@ -1565,7 +1565,7 @@ def write_workflow_prov(graph, filename=None, format="all"):
         _, hashval, _, _ = node.hash_exists()
         attrs = {
             pm.PROV["type"]: nipype_ns[classname],
-            pm.PROV["label"]: "_".join((classname, node.name)),
+            pm.PROV["label"]: f"{classname}_{node.name}",
             nipype_ns["hashval"]: hashval,
         }
         process = ps.g.activity(get_id(), None, None, attrs)

nipype/pipeline/engine/workflows.py

@@ -394,10 +394,7 @@ def list_node_names(self):
         for node in nx.topological_sort(self._graph):
             if isinstance(node, Workflow):
                 outlist.extend(
-                    [
-                        ".".join((node.name, nodename))
-                        for nodename in node.list_node_names()
-                    ]
+                    f"{node.name}.{nodename}" for nodename in node.list_node_names()
                 )
             else:
                 outlist.append(node.name)
@@ -931,7 +928,7 @@ def _reset_hierarchy(self):
             if isinstance(node, Workflow):
                 node._reset_hierarchy()
                 for innernode in node._graph.nodes():
-                    innernode._hierarchy = ".".join((self.name, innernode._hierarchy))
+                    innernode._hierarchy = f"{self.name}.{innernode._hierarchy}"
             else:
                 node._hierarchy = self.name
 
@@ -995,7 +992,7 @@ def _generate_flatgraph(self):
                 # logger.debug('expanding workflow: %s', node)
                 node._generate_flatgraph()
                 for innernode in node._graph.nodes():
-                    innernode._hierarchy = ".".join((self.name, innernode._hierarchy))
+                    innernode._hierarchy = f"{self.name}.{innernode._hierarchy}"
                 self._graph.add_nodes_from(node._graph.nodes())
                 self._graph.add_edges_from(node._graph.edges(data=True))
         if nodes2remove:

nipype/pipeline/plugins/tests/test_legacymultiproc_nondaemon.py

@@ -122,7 +122,7 @@ def run_multiproc_nondaemon_with_flag(nondaemon_flag):
         plugin_args={"n_procs": 2, "non_daemon": nondaemon_flag},
     )
 
-    names = [".".join((node._hierarchy, node.name)) for node in execgraph.nodes()]
+    names = [f"{node._hierarchy}.{node.name}" for node in execgraph.nodes()]
     node = list(execgraph.nodes())[names.index("pipe.f2")]
     result = node.get_output("sum_out")
     os.chdir(cur_dir)

nipype/pipeline/plugins/tests/test_linear.py

@@ -39,7 +39,7 @@ def test_run_in_series(tmpdir):
     pipe.base_dir = os.getcwd()
     mod1.inputs.input1 = 1
     execgraph = pipe.run(plugin="Linear")
-    names = [".".join((node._hierarchy, node.name)) for node in execgraph.nodes()]
+    names = [f"{node._hierarchy}.{node.name}" for node in execgraph.nodes()]
     node = list(execgraph.nodes())[names.index("pipe.mod1")]
     result = node.get_output("output1")
     assert result == [1, 1]

nipype/pipeline/plugins/tests/test_oar.py

@@ -38,7 +38,7 @@ def test_run_oargraph(tmp_path):
     pipe.base_dir = os.getcwd()
     mod1.inputs.input1 = 1
     execgraph = pipe.run(plugin="OAR")
-    names = [".".join((node._hierarchy, node.name)) for node in execgraph.nodes()]
+    names = [f"{node._hierarchy}.{node.name}" for node in execgraph.nodes()]
     node = list(execgraph.nodes())[names.index("pipe.mod1")]
     result = node.get_output("output1")
     assert result == [1, 1]

nipype/pipeline/plugins/tests/test_pbs.py

@@ -37,7 +37,7 @@ def test_run_pbsgraph(tmp_path):
     pipe.connect([(mod1, mod2, [("output1", "input1")])])
     mod1.inputs.input1 = 1
     execgraph = pipe.run(plugin="PBSGraph")
-    names = [".".join((node._hierarchy, node.name)) for node in execgraph.nodes()]
+    names = [f"{node._hierarchy}.{node.name}" for node in execgraph.nodes()]
     node = list(execgraph.nodes())[names.index("pipe.mod1")]
     result = node.get_output("output1")
     assert result == [1, 1]

nipype/pipeline/plugins/tests/test_somaflow.py

@@ -42,7 +42,7 @@ def test_run_somaflow(tmpdir):
     pipe.base_dir = os.getcwd()
     mod1.inputs.input1 = 1
     execgraph = pipe.run(plugin="SomaFlow")
-    names = [".".join((node._hierarchy, node.name)) for node in execgraph.nodes()]
+    names = [f"{node._hierarchy}.{node.name}" for node in execgraph.nodes()]
     node = list(execgraph.nodes())[names.index("pipe.mod1")]
     result = node.get_output("output1")
     assert result == [1, 1]

nipype/pipeline/plugins/tools.py

@@ -83,10 +83,8 @@ def report_nodes_not_run(notrun):
     if notrun:
         logger.info("***********************************")
         for info in notrun:
-            logger.error(
-                "could not run node: %s"
-                % ".".join((info["node"]._hierarchy, info["node"]._id))
-            )
+            node = info["node"]
+            logger.error(f"could not run node: {node._hierarchy}.{node._id}")
             logger.info("crashfile: %s" % info["crashfile"])
             logger.debug("The following dependent nodes were not run")
             for subnode in info["dependents"]:

nipype/utils/docparse.py

@@ -118,7 +118,7 @@ def format_params(paramlist, otherlist=None):
         otherlist.insert(0, hdr)
         otherlist.insert(0, "\n")
         otherparams = "\n".join(otherlist)
-        doc = "".join([doc, otherparams])
+        doc = f"{doc}{otherparams}"
     return doc
 
 
@@ -251,7 +251,7 @@ def get_doc(cmd, opt_map, help_flag=None, trap_error=True):
     if cmd_path == "":
         raise Exception("Command %s not found" % cmd.split(" ")[0])
     if help_flag:
-        cmd = " ".join((cmd, help_flag))
+        cmd = f"{cmd} {help_flag}"
     doc = grab_doc(cmd, trap_error)
     opts = reverse_opt_map(opt_map)
     return build_doc(doc, opts)
@@ -332,7 +332,7 @@ def get_params_from_doc(cmd, style="--", help_flag=None, trap_error=True):
     if cmd_path == "":
         raise Exception("Command %s not found" % cmd.split(" ")[0])
     if help_flag:
-        cmd = " ".join((cmd, help_flag))
+        cmd = f"{cmd} {help_flag}"
     doc = grab_doc(cmd, trap_error)
     return _parse_doc(doc, style)
 

nipype/utils/filemanip.py

@@ -137,7 +137,7 @@ def hash_rename(filename, hashvalue):
     and sets path to output_directory
     """
     path, name, ext = split_filename(filename)
-    newfilename = "".join((name, "_0x", hashvalue, ext))
+    newfilename = f"{name}_0x{hashvalue}{ext}"
     return op.join(path, newfilename)
 
 

nipype/utils/tests/test_filemanip.py

@@ -87,7 +87,7 @@ def test_hash_rename(filename, newname):
 def test_check_forhash():
     fname = "foobar"
     orig_hash = "_0x4323dbcefdc51906decd8edcb3327943"
-    hashed_name = "".join((fname, orig_hash, ".nii"))
+    hashed_name = f"{fname}{orig_hash}.nii"
     result, hash = check_forhash(hashed_name)
     assert result
     assert hash == [orig_hash]

tools/checkspecs.py

@@ -143,7 +143,7 @@ def _parse_lines(self, linesource, module):
                 # exclude private stuff
                 name = self._get_object_name(line)
                 if not name.startswith("_") and self._survives_exclude(
-                    ".".join((module, name)), "class"
+                    f"{module}.{name}", "class"
                 ):
                     classes.append(name)
             else:
@@ -449,7 +449,7 @@ def discover_modules(self):
             # Check directory names for packages
             root_uri = self._path2uri(os.path.join(self.root_path, dirpath))
             for dirname in dirnames[:]:  # copy list - we modify inplace
-                package_uri = ".".join((root_uri, dirname))
+                package_uri = f"{root_uri}.{dirname}"
                 if self._uri2path(package_uri) and self._survives_exclude(
                     package_uri, "package"
                 ):
@@ -459,7 +459,7 @@ def discover_modules(self):
             # Check filenames for modules
             for filename in filenames:
                 module_name = filename[:-3]
-                module_uri = ".".join((root_uri, module_name))
+                module_uri = f"{root_uri}.{module_name}"
                 if self._uri2path(module_uri) and self._survives_exclude(
                     module_uri, "module"
                 ):