14141621866

Committed 29 Mar 2025 03:13AM UTC coverage: 74.142% (-0.009%) from 74.151%

Build # 14141621866

Build Type

push

github

Committed by

web-flow

Commit Message

[bugfix] Fix errors for scripts with spaces (backport #6027) (#6036)

Co-authored-by: Hailin Wang <hailin.wang@connect.polyu.hk>

Run Details

0 of 4 new or added lines in 1 file covered. (0.0%)

2 existing lines in 1 file now uncovered.

24550 of 33112 relevant lines covered (74.14%)

0.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

48.53

/src/abqpy/cli.py

from __future__ import annotations

import os
import subprocess

from typeguard import typechecked
from typing_extensions import Self

from .config import config


@typechecked
class AbqpyCLIBase:
    """Base class for Abaqus/CAE command line interface to run Abaqus commands."""

    def _parse_options(self, **options: str | int | bool | None) -> str:
        """Parse options to be passed to Abaqus/CAE command line interface.

        If the value is a string or an integer, the option will be passed as ``option=value``; if the value is a
        boolean, the option will be passed as ``option`` if True, or ignored if False; if the value is None, the option
        will be ignored.
        """
        return " ".join([f"{k}={v}" if isinstance(v, (str, int)) and not isinstance(v, bool) else
                         k for k, v in options.items() if v])  # fmt: skip

    def run(self, cmd: str):
        """Run custom command."""
        cmd = cmd.strip()
        message = f"Running the following command: {cmd}"
        print("", "-" * len(message), message, "-" * len(message), sep="\n")
        if config.execution_method == "os":
            return os.system(cmd)
        elif config.execution_method == "subprocess":
            return subprocess.run(cmd, shell=True, stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
        else:
            raise ValueError(f"Invalid execution method: {config.execution_method}")

    def abaqus(self, *args, **options):
        """Run custom Abaqus command: ``abaqus {args} {options}``, arguments are separated by space, options are
        handled by the :meth:`._parse_options` method.

        Parameters
        ----------
        args, options
            Arguments and options to be passed to the Abaqus command.
        """
        abaqus = os.environ.get("ABAQUS_BAT_PATH", "abaqus")
        args, options = " ".join(args), self._parse_options(**options)
        return self.run(abaqus + (f" {args}" if args else "") + (f" {options}" if options else ""))


@typechecked
class AbqpyCLI(AbqpyCLIBase):
    """The abqpy command line interface."""

    @property
    def misc(self) -> Self:
        """Miscellaneous commands for backward compatibility."""
        return self

    def cae(
        self,
        script: str,
        *args,
        database: str | None = None,
        replay: str | None = None,
        recover: str | None = None,
        startup: str | None = None,
        gui: bool = False,
        envstartup: bool = True,
        savedOptions: bool = True,
        savedGuiPrefs: bool = True,
        startupDialog: bool = True,
        custom: str | None = None,
        guiTester: str | None = None,
        guiRecord: bool | None = None,
    ):
        """Run Abaqus/CAE command.

        Parameters
        ----------
        script : str
            The name of the python script to run
        args : str
            Extra arguments to be passed after the Abaqus/CAE command line options
        database : str, optional
            The name of the database file to open, by default None
        replay : str, optional
            The name of the replay file to open, by default None
        recover : str, optional
            The name of the journal file to open, by default None
        startup : str, optional
            The name of the startup file to open, by default None
        gui : bool, optional
            Run Abaqus/CAE command with the graphical user interface (GUI mode), by default False.
        envstartup : bool, optional
            Execute the Abaqus/CAE startup file, by default True
        savedOptions : bool, optional
            Use the saved options, by default True
        savedGuiPrefs : bool, optional
            Use the saved GUI preferences, by default True
        startupDialog : bool, optional
            Display the startup dialog, by default True
        custom : str, optional
            The name of the file containing Abaqus GUI Toolkit commands to be executed, by default None
        guiTester : str, optional
            This option starts a separate user interface containing the Abaqus Python development environment along
            with Abaqus/CAE.
        guiRecord : bool, optional
            Record the GUI commands to a file, by default None
        """
        # Parse options
        script = f'"{script}"'.replace('""', '"') if " " in script else script
        options = self._parse_options(script=script if gui else None, noGUI=script if not gui else None,
                                      database=database, replay=replay, recover=recover, startup=startup,
                                      noenvstartup=not envstartup, noSavedOptions=not savedOptions,
                                      noSavedGuiPrefs=not savedGuiPrefs, noStartupDialog=not startupDialog,
                                      custom=custom, guiTester=guiTester,
                                      guiRecord=True if guiRecord is True else None,
                                      guiNoRecord=True if guiRecord is False else None)  # fmt: skip
        args = ("--", *args) if args else ()

        # Execute command
        return self.abaqus("cae", options, *args)

    viewer = cae

    def pde(self, *scripts: str, script: str | None = None, **options):
        """Run Abaqus/PDE command.

        Parameters
        ----------
        scripts : str
            Scripts to be included in the Abaqus/PDE session
        script : str, optional
            Script to be opened at startup in Abaqus/PDE, by default None
        options
            Abaqus/CAE command line arguments
        """
        script = f'"{script}"'.replace('""', '"') if script is not None and " " in script else script
        scripts = tuple(f'"{s}"'.replace('""', '"') if " " in s else s for s in scripts)
        cae_opts = self._parse_options(**options)
        args = (*scripts,) + ((f"script={script}",) if script else ()) + ("-pde",) + ((cae_opts,) if cae_opts else ())
        return self.abaqus("pde", *args)

    def python(
        self,
        script: str,
        *args,
        sim: str | None = None,
        log: str | None = None,
    ):
        """Run Abaqus/Python command.

        Parameters
        ----------
        script : str
            The name of the python script to run
        args : str
            Extra arguments to be passed after the Abaqus/CAE command line options
        sim : str, optional
            The name of the simulation file to open, by default None
        log : str, optional
            The name of the log file to open, by default None
        """
        # Parse options
        options = self._parse_options(sim=sim, log=log)

        # Execute command
        script = f'"{script}"'.replace('""', '"') if " " in script else script
        return self.abaqus("python", script, options, *args)

    @typechecked
    def optimization(
        self,
        task: str,
        job: str,
        *,
        cpus: int | None = None,
        gpus: int | None = None,
        memory: int | None = None,
        interactive: bool = False,
        globalmodel: str | None = None,
        scratch: str | None = None,
    ):
        """Run Abaqus optimization command.

        Parameters
        ----------
        task : str
            The file containing the parameters that are used to execute the optimization.
        job : str
            The name of the folder in which the results of the optimization are stored.
        cpus : int, optional
            The number of processors to use during an analysis run if parallel processing is available.
        gpus : int, optional
            This acceleration of the Abaqus/Standard direct solver.
        memory : int, optional
            Maximum amount of memory or maximum percentage of the physical memory that can be allocated.
        interactive : bool, optional
            This option will cause the job to run interactively.
        globalmodel : str, optional
            The name of the global model's results file, ODB output database file, or SIM database file.
        scratch : str, optional
            The name of the directory used for scratch files.
        """
        # Execute command
        return self.abaqus("optimization", task=task, job=job, cpus=cpus, gpus=gpus, memory=memory,
                           interactive=interactive, globalmodel=globalmodel, scratch=scratch)  # fmt: skip

    def help(self, *args, **options):
        return self.abaqus("help", *args, **options)

    def information(self, *args, **options):
        return self.abaqus("information", *args, **options)

    def whereami(self, *args, **options):
        return self.abaqus("whereami", *args, **options)

    def cse(self, *args, **options):
        return self.abaqus("cse", *args, **options)

    def cosimulation(self, *args, **options):
        return self.abaqus("cosimulation", *args, **options)

    def fmu(self, *args, **options):
        return self.abaqus("fmu", *args, **options)

    def script(self, *args, **options):
        return self.abaqus("script", *args, **options)

    def doc(self, *args, **options):
        return self.abaqus("doc", *args, **options)

    def licensing(self, *args, **options):
        return self.abaqus("licensing", *args, **options)

    def ascfil(self, *args, **options):
        return self.abaqus("ascfil", *args, **options)

    def append(self, *args, **options):
        return self.abaqus("append", *args, **options)

    def findkeyword(self, *args, **options):
        return self.abaqus("findkeyword", *args, **options)

    def fetch(self, *args, **options):
        return self.abaqus("fetch", *args, **options)

    def make(self, *args, **options):
        return self.abaqus("make", *args, **options)

    def upgrade(self, *args, **options):
        return self.abaqus("upgrade", *args, **options)

    def sim_version(self, *args, **options):
        return self.abaqus("sim_version", *args, **options)

    def odb2sim(self, *args, **options):
        return self.abaqus("odb2sim", *args, **options)

    def odbreport(self, *args, **options):
        return self.abaqus("odbReport", *args, **options)

    def restartjoin(self, *args, **options):
        return self.abaqus("restartjoin", *args, **options)

    def substructurecombine(self, *args, **options):
        return self.abaqus("substructurecombine", *args, **options)

    def substructurerecover(self, *args, **options):
        return self.abaqus("substructurerecover", *args, **options)

    def odbcombine(self, *args, **options):
        return self.abaqus("odbcombine", *args, **options)

    def networkDBConnector(self, *args, **options):
        return self.abaqus("networkDBConnector", *args, **options)

    def emloads(self, *args, **options):
        return self.abaqus("emloads", *args, **options)

    def mtxasm(self, *args, **options):
        return self.abaqus("mtxasm", *args, **options)

    def fromnastran(self, *args, **options):
        return self.abaqus("fromnastran", *args, **options)

    def tonastran(self, *args, **options):
        return self.abaqus("tonastran", *args, **options)

    def fromansys(self, *args, **options):
        return self.abaqus("fromansys", *args, **options)

    def frompamcrash(self, *args, **options):
        return self.abaqus("frompamcrash", *args, **options)

    def fromradioss(self, *args, **options):
        return self.abaqus("fromradioss", *args, **options)

    def toOutput2(self, *args, **options):
        return self.abaqus("toOutput2", *args, **options)

    def fromdyna(self, *args, **options):
        return self.abaqus("fromdyna", *args, **options)

    def tozaero(self, *args, **options):
        return self.abaqus("tozaero", *args, **options)

    def adams(self, *args, **options):
        return self.abaqus("adams", *args, **options)

    def tosimpack(self, *args, **options):
        return self.abaqus("tosimpack", *args, **options)

    def fromsimpack(self, *args, **options):
        return self.abaqus("fromsimpack", *args, **options)

    def toexcite(self, *args, **options):
        return self.abaqus("toexcite", *args, **options)

    def moldflow(self, *args, **options):
        return self.abaqus("moldflow", *args, **options)

    def encrypt(self, *args, **options):
        return self.abaqus("encrypt", *args, **options)

    def decrypt(self, *args, **options):
        return self.abaqus("decrypt", *args, **options)

    def suspend(self, *args, **options):
        return self.abaqus("suspend", *args, **options)

    def resume(self, *args, **options):
        return self.abaqus("resume", *args, **options)

    def terminate(self, *args, **options):
        return self.abaqus("terminate", *args, **options)

    def sysVerify(self, *args, **options):
        return self.abaqus("sysVerify", *args, **options)


#: The abqpy command line interface, use this object to run abqpy commands from the python scripts
abaqus = AbqpyCLI()

1	from __future__ import annotations	1✔
2
3	import os	1✔
4	import subprocess	1✔
5
6	from typeguard import typechecked	1✔
7	from typing_extensions import Self	1✔
8
9	from .config import config	1✔
10
11
12	@typechecked	1✔
13	class AbqpyCLIBase:	1✔
14	"""Base class for Abaqus/CAE command line interface to run Abaqus commands."""
15
16	def _parse_options(self, **options: str \| int \| bool \| None) -> str:	1✔
17	"""Parse options to be passed to Abaqus/CAE command line interface.
18
19	If the value is a string or an integer, the option will be passed as ``option=value``; if the value is a
20	boolean, the option will be passed as ``option`` if True, or ignored if False; if the value is None, the option
21	will be ignored.
22	"""
23	return " ".join([f"{k}={v}" if isinstance(v, (str, int)) and not isinstance(v, bool) else	×
24	k for k, v in options.items() if v]) # fmt: skip
25
26	def run(self, cmd: str):	1✔
27	"""Run custom command."""
28	cmd = cmd.strip()	×
29	message = f"Running the following command: {cmd}"	×
30	print("", "-" * len(message), message, "-" * len(message), sep="\n")	×
31	if config.execution_method == "os":	×
32	return os.system(cmd)	×
33	elif config.execution_method == "subprocess":	×
34	return subprocess.run(cmd, shell=True, stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True)	×
35	else:
36	raise ValueError(f"Invalid execution method: {config.execution_method}")	×
37
38	def abaqus(self, args, *options):	1✔
39	"""Run custom Abaqus command: ``abaqus {args} {options}``, arguments are separated by space, options are
40	handled by the :meth:`._parse_options` method.
41
42	Parameters
43	----------
44	args, options
45	Arguments and options to be passed to the Abaqus command.
46	"""
47	abaqus = os.environ.get("ABAQUS_BAT_PATH", "abaqus")	×
48	args, options = " ".join(args), self._parse_options(**options)	×
49	return self.run(abaqus + (f" {args}" if args else "") + (f" {options}" if options else ""))	×
50
51
52	@typechecked	1✔
53	class AbqpyCLI(AbqpyCLIBase):	1✔
54	"""The abqpy command line interface."""
55
56	@property	1✔
57	def misc(self) -> Self:	1✔
58	"""Miscellaneous commands for backward compatibility."""
59	return self	×
60
61	def cae(	1✔
62	self,
63	script: str,
64	*args,
65	database: str \| None = None,
66	replay: str \| None = None,
67	recover: str \| None = None,
68	startup: str \| None = None,
69	gui: bool = False,
70	envstartup: bool = True,
71	savedOptions: bool = True,
72	savedGuiPrefs: bool = True,
73	startupDialog: bool = True,
74	custom: str \| None = None,
75	guiTester: str \| None = None,
76	guiRecord: bool \| None = None,
77	):
78	"""Run Abaqus/CAE command.
79
80	Parameters
81	----------
82	script : str
83	The name of the python script to run
84	args : str
85	Extra arguments to be passed after the Abaqus/CAE command line options
86	database : str, optional
87	The name of the database file to open, by default None
88	replay : str, optional
89	The name of the replay file to open, by default None
90	recover : str, optional
91	The name of the journal file to open, by default None
92	startup : str, optional
93	The name of the startup file to open, by default None
94	gui : bool, optional
95	Run Abaqus/CAE command with the graphical user interface (GUI mode), by default False.
96	envstartup : bool, optional
97	Execute the Abaqus/CAE startup file, by default True
98	savedOptions : bool, optional
99	Use the saved options, by default True
100	savedGuiPrefs : bool, optional
101	Use the saved GUI preferences, by default True
102	startupDialog : bool, optional
103	Display the startup dialog, by default True
104	custom : str, optional
105	The name of the file containing Abaqus GUI Toolkit commands to be executed, by default None
106	guiTester : str, optional
107	This option starts a separate user interface containing the Abaqus Python development environment along
108	with Abaqus/CAE.
109	guiRecord : bool, optional
110	Record the GUI commands to a file, by default None
111	"""
112	# Parse options
NEW 113	script = f'"{script}"'.replace('""', '"') if " " in script else script	×
UNCOV 114	options = self._parse_options(script=script if gui else None, noGUI=script if not gui else None,	×
115	database=database, replay=replay, recover=recover, startup=startup,
116	noenvstartup=not envstartup, noSavedOptions=not savedOptions,
117	noSavedGuiPrefs=not savedGuiPrefs, noStartupDialog=not startupDialog,
118	custom=custom, guiTester=guiTester,
119	guiRecord=True if guiRecord is True else None,
120	guiNoRecord=True if guiRecord is False else None) # fmt: skip
121	args = ("--", *args) if args else ()	×
122
123	# Execute command
124	return self.abaqus("cae", options, *args)	×
125
126	viewer = cae	1✔
127
128	def pde(self, scripts: str, script: str \| None = None, *options):	1✔
129	"""Run Abaqus/PDE command.
130
131	Parameters
132	----------
133	scripts : str
134	Scripts to be included in the Abaqus/PDE session
135	script : str, optional
136	Script to be opened at startup in Abaqus/PDE, by default None
137	options
138	Abaqus/CAE command line arguments
139	"""
NEW 140	script = f'"{script}"'.replace('""', '"') if script is not None and " " in script else script	×
NEW 141	scripts = tuple(f'"{s}"'.replace('""', '"') if " " in s else s for s in scripts)	×
142	cae_opts = self._parse_options(**options)	×
143	args = (*scripts,) + ((f"script={script}",) if script else ()) + ("-pde",) + ((cae_opts,) if cae_opts else ())	×
144	return self.abaqus("pde", *args)	×
145
146	def python(	1✔
147	self,
148	script: str,
149	*args,
150	sim: str \| None = None,
151	log: str \| None = None,
152	):
153	"""Run Abaqus/Python command.
154
155	Parameters
156	----------
157	script : str
158	The name of the python script to run
159	args : str
160	Extra arguments to be passed after the Abaqus/CAE command line options
161	sim : str, optional
162	The name of the simulation file to open, by default None
163	log : str, optional
164	The name of the log file to open, by default None
165	"""
166	# Parse options
167	options = self._parse_options(sim=sim, log=log)	×
168
169	# Execute command
NEW 170	script = f'"{script}"'.replace('""', '"') if " " in script else script	×
UNCOV 171	return self.abaqus("python", script, options, *args)	×
172
173	@typechecked	1✔
174	def optimization(	1✔
175	self,
176	task: str,
177	job: str,
178	*,
179	cpus: int \| None = None,
180	gpus: int \| None = None,
181	memory: int \| None = None,
182	interactive: bool = False,
183	globalmodel: str \| None = None,
184	scratch: str \| None = None,
185	):
186	"""Run Abaqus optimization command.
187
188	Parameters
189	----------
190	task : str
191	The file containing the parameters that are used to execute the optimization.
192	job : str
193	The name of the folder in which the results of the optimization are stored.
194	cpus : int, optional
195	The number of processors to use during an analysis run if parallel processing is available.
196	gpus : int, optional
197	This acceleration of the Abaqus/Standard direct solver.
198	memory : int, optional
199	Maximum amount of memory or maximum percentage of the physical memory that can be allocated.
200	interactive : bool, optional
201	This option will cause the job to run interactively.
202	globalmodel : str, optional
203	The name of the global model's results file, ODB output database file, or SIM database file.
204	scratch : str, optional
205	The name of the directory used for scratch files.
206	"""
207	# Execute command
208	return self.abaqus("optimization", task=task, job=job, cpus=cpus, gpus=gpus, memory=memory,	×
209	interactive=interactive, globalmodel=globalmodel, scratch=scratch) # fmt: skip
210
211	def help(self, args, *options):	1✔
212	return self.abaqus("help", args, *options)	×
213
214	def information(self, args, *options):	1✔
215	return self.abaqus("information", args, *options)	×
216
217	def whereami(self, args, *options):	1✔
218	return self.abaqus("whereami", args, *options)	×
219
220	def cse(self, args, *options):	1✔
221	return self.abaqus("cse", args, *options)	×
222
223	def cosimulation(self, args, *options):	1✔
224	return self.abaqus("cosimulation", args, *options)	×
225
226	def fmu(self, args, *options):	1✔
227	return self.abaqus("fmu", args, *options)	×
228
229	def script(self, args, *options):	1✔
230	return self.abaqus("script", args, *options)	×
231
232	def doc(self, args, *options):	1✔
233	return self.abaqus("doc", args, *options)	×
234
235	def licensing(self, args, *options):	1✔
236	return self.abaqus("licensing", args, *options)	×
237
238	def ascfil(self, args, *options):	1✔
239	return self.abaqus("ascfil", args, *options)	×
240
241	def append(self, args, *options):	1✔
242	return self.abaqus("append", args, *options)	×
243
244	def findkeyword(self, args, *options):	1✔
245	return self.abaqus("findkeyword", args, *options)	×
246
247	def fetch(self, args, *options):	1✔
248	return self.abaqus("fetch", args, *options)	×
249
250	def make(self, args, *options):	1✔
251	return self.abaqus("make", args, *options)	×
252
253	def upgrade(self, args, *options):	1✔
254	return self.abaqus("upgrade", args, *options)	×
255
256	def sim_version(self, args, *options):	1✔
257	return self.abaqus("sim_version", args, *options)	×
258
259	def odb2sim(self, args, *options):	1✔
260	return self.abaqus("odb2sim", args, *options)	×
261
262	def odbreport(self, args, *options):	1✔
263	return self.abaqus("odbReport", args, *options)	×
264
265	def restartjoin(self, args, *options):	1✔
266	return self.abaqus("restartjoin", args, *options)	×
267
268	def substructurecombine(self, args, *options):	1✔
269	return self.abaqus("substructurecombine", args, *options)	×
270
271	def substructurerecover(self, args, *options):	1✔
272	return self.abaqus("substructurerecover", args, *options)	×
273
274	def odbcombine(self, args, *options):	1✔
275	return self.abaqus("odbcombine", args, *options)	×
276
277	def networkDBConnector(self, args, *options):	1✔
278	return self.abaqus("networkDBConnector", args, *options)	×
279
280	def emloads(self, args, *options):	1✔
281	return self.abaqus("emloads", args, *options)	×
282
283	def mtxasm(self, args, *options):	1✔
284	return self.abaqus("mtxasm", args, *options)	×
285
286	def fromnastran(self, args, *options):	1✔
287	return self.abaqus("fromnastran", args, *options)	×
288
289	def tonastran(self, args, *options):	1✔
290	return self.abaqus("tonastran", args, *options)	×
291
292	def fromansys(self, args, *options):	1✔
293	return self.abaqus("fromansys", args, *options)	×
294
295	def frompamcrash(self, args, *options):	1✔
296	return self.abaqus("frompamcrash", args, *options)	×
297
298	def fromradioss(self, args, *options):	1✔
299	return self.abaqus("fromradioss", args, *options)	×
300
301	def toOutput2(self, args, *options):	1✔
302	return self.abaqus("toOutput2", args, *options)	×
303
304	def fromdyna(self, args, *options):	1✔
305	return self.abaqus("fromdyna", args, *options)	×
306
307	def tozaero(self, args, *options):	1✔
308	return self.abaqus("tozaero", args, *options)	×
309
310	def adams(self, args, *options):	1✔
311	return self.abaqus("adams", args, *options)	×
312
313	def tosimpack(self, args, *options):	1✔
314	return self.abaqus("tosimpack", args, *options)	×
315
316	def fromsimpack(self, args, *options):	1✔
317	return self.abaqus("fromsimpack", args, *options)	×
318
319	def toexcite(self, args, *options):	1✔
320	return self.abaqus("toexcite", args, *options)	×
321
322	def moldflow(self, args, *options):	1✔
323	return self.abaqus("moldflow", args, *options)	×
324
325	def encrypt(self, args, *options):	1✔
326	return self.abaqus("encrypt", args, *options)	×
327
328	def decrypt(self, args, *options):	1✔
329	return self.abaqus("decrypt", args, *options)	×
330
331	def suspend(self, args, *options):	1✔
332	return self.abaqus("suspend", args, *options)	×
333
334	def resume(self, args, *options):	1✔
335	return self.abaqus("resume", args, *options)	×
336
337	def terminate(self, args, *options):	1✔
338	return self.abaqus("terminate", args, *options)	×
339
340	def sysVerify(self, args, *options):	1✔
341	return self.abaqus("sysVerify", args, *options)	×
342
343
344	#: The abqpy command line interface, use this object to run abqpy commands from the python scripts
345	abaqus = AbqpyCLI()	1✔

haiiliin / abqpy / 14141621866

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous