]>
git.ipfire.org Git - thirdparty/u-boot.git/blob - tools/u_boot_pylib/command.py
1 # SPDX-License-Identifier: GPL-2.0+
2 # Copyright (c) 2011 The Chromium OS Authors.
7 from u_boot_pylib
import cros_subprocess
9 """Shell command ease-ups for Python."""
12 """A class which captures the result of executing a command.
15 stdout: stdout obtained from command, as a string
16 stderr: stderr obtained from command, as a string
17 return_code: Return code from command
18 exception: Exception received, or None if all ok
20 def __init__(self
, stdout
='', stderr
='', combined
='', return_code
=0,
24 self
.combined
= combined
25 self
.return_code
= return_code
26 self
.exception
= exception
28 def to_output(self
, binary
):
30 self
.stdout
= self
.stdout
.decode('utf-8')
31 self
.stderr
= self
.stderr
.decode('utf-8')
32 self
.combined
= self
.combined
.decode('utf-8')
36 # This permits interception of RunPipe for test purposes. If it is set to
37 # a function, then that function is called with the pipe list being
38 # executed. Otherwise, it is assumed to be a CommandResult object, and is
39 # returned as the result for every run_pipe() call.
40 # When this value is None, commands are executed as normal.
43 def run_pipe(pipe_list
, infile
=None, outfile
=None,
44 capture
=False, capture_stderr
=False, oneline
=False,
45 raise_on_error
=True, cwd
=None, binary
=False,
46 output_func
=None, **kwargs
):
48 Perform a command pipeline, with optional input/output filenames.
51 pipe_list: List of command lines to execute. Each command line is
52 piped into the next, and is itself a list of strings. For
53 example [ ['ls', '.git'] ['wc'] ] will pipe the output of
55 infile: File to provide stdin to the pipeline
56 outfile: File to store stdout
57 capture: True to capture output
58 capture_stderr: True to capture stderr
59 oneline: True to strip newline chars from output
60 output_func: Output function to call with each output fragment
61 (if it returns True the function terminates)
62 kwargs: Additional keyword arguments to cros_subprocess.Popen()
67 if hasattr(test_result
, '__call__'):
68 # pylint: disable=E1102
69 result
= test_result(pipe_list
=pipe_list
)
74 # No result: fall through to normal processing
75 result
= CommandResult(b
'', b
'', b
'')
77 pipeline
= list(pipe_list
)
78 user_pipestr
= '|'.join([' '.join(pipe
) for pipe
in pipe_list
])
79 kwargs
['stdout'] = None
80 kwargs
['stderr'] = None
83 if last_pipe
is not None:
84 kwargs
['stdin'] = last_pipe
.stdout
86 kwargs
['stdin'] = open(infile
, 'rb')
87 if pipeline
or capture
:
88 kwargs
['stdout'] = cros_subprocess
.PIPE
90 kwargs
['stdout'] = open(outfile
, 'wb')
92 kwargs
['stderr'] = cros_subprocess
.PIPE
95 last_pipe
= cros_subprocess
.Popen(cmd
, cwd
=cwd
, **kwargs
)
96 except Exception as err
:
97 result
.exception
= err
99 raise Exception("Error running '%s': %s" % (user_pipestr
, str))
100 result
.return_code
= 255
101 return result
.to_output(binary
)
104 result
.stdout
, result
.stderr
, result
.combined
= (
105 last_pipe
.communicate_filter(output_func
))
106 if result
.stdout
and oneline
:
107 result
.output
= result
.stdout
.rstrip(b
'\r\n')
108 result
.return_code
= last_pipe
.wait()
110 result
.return_code
= os
.waitpid(last_pipe
.pid
, 0)[1]
111 if raise_on_error
and result
.return_code
:
112 raise Exception("Error running '%s'" % user_pipestr
)
113 return result
.to_output(binary
)
115 def output(*cmd
, **kwargs
):
116 kwargs
['raise_on_error'] = kwargs
.get('raise_on_error', True)
117 return run_pipe([cmd
], capture
=True, **kwargs
).stdout
119 def output_one_line(*cmd
, **kwargs
):
120 """Run a command and output it as a single-line string
122 The command us expected to produce a single line of output
125 String containing output of command
127 raise_on_error
= kwargs
.pop('raise_on_error', True)
128 result
= run_pipe([cmd
], capture
=True, oneline
=True,
129 raise_on_error
=raise_on_error
, **kwargs
).stdout
.strip()
132 def run(*cmd
, **kwargs
):
133 return run_pipe([cmd
], **kwargs
).stdout
136 return run_pipe([cmd
], capture
=True).stdout
139 cros_subprocess
.stay_alive
= False