cardinal_pythonlib.subproc
Original code copyright (C) 2009-2022 Rudolf Cardinal (rudolf@pobox.com).
This file is part of cardinal_pythonlib.
Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at
Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.
- class cardinal_pythonlib.subproc.AsynchronousFileReader(fd: BinaryIO, queue: Queue, encoding: str, line_terminators: List[str] | None = None, cmdargs: List[str] | None = None, suppress_decoding_errors: bool = True)[source]
Helper class to implement asynchronous reading of a file in a separate thread. Pushes read lines on a queue to be consumed in another thread.
Modified from https://stefaanlippens.net/python-asynchronous-subprocess-pipe-reading/.
- Parameters:
fd¶ – file-like object to read from
queue¶ – queue to write to
encoding¶ – encoding to use when reading from the file
line_terminators¶ – valid line terminators
cmdargs¶ – for display purposes only: command that produced/is producing the file-like object
suppress_decoding_errors¶ – trap any
UnicodeDecodeError
?
- cardinal_pythonlib.subproc.check_call_process(args: List[str]) None [source]
Logs the command arguments, then executes the command via
subprocess.check_call()
.
- cardinal_pythonlib.subproc.check_call_verbose(args: List[str], log_level: int | None = 20, **kwargs) None [source]
Prints a copy/paste-compatible version of a command, then runs it.
- cardinal_pythonlib.subproc.fail() NoReturn [source]
Call when a child process has failed, and this will print an error message to
stdout
and executesys.exit(1)
(which will, in turn, call anyatexit
handler to kill children of this process).
- cardinal_pythonlib.subproc.kill_child_processes() None [source]
Kills children of this process that were registered in the
processes
variable.Use with
@atexit.register
.
- cardinal_pythonlib.subproc.mimic_user_input(args: List[str], source_challenge_response: List[Tuple[SubprocSource, str, str | SubprocCommand]], line_terminators: List[str] | None = None, print_stdout: bool = False, print_stderr: bool = False, print_stdin: bool = False, stdin_encoding: str | None = None, stdout_encoding: str | None = None, suppress_decoding_errors: bool = True, sleep_time_s: float = 0.1) None [source]
Run an external command. Pretend to be a human by sending text to the subcommand (responses) when the external command sends us triggers (challenges).
This is a bit nasty.
- Parameters:
args¶ – command-line arguments
source_challenge_response¶ – list of tuples of the format
(challsrc, challenge, response)
; see belowline_terminators¶ – valid line terminators
print_stdout¶ –
print_stderr¶ –
print_stdin¶ –
stdin_encoding¶ –
stdout_encoding¶ –
suppress_decoding_errors¶ – trap any
UnicodeDecodeError
?sleep_time_s¶ –
The
(challsrc, challenge, response)
tuples have this meaning:challsrc
: where is the challenge coming from? Must be one of the objectsSOURCE_STDOUT
orSOURCE_STDERR
;challenge
: text of challengeresponse
: text of response (send to the subcommand’sstdin
).
Example (modified from
CorruptedZipReader
):from cardinal_pythonlib.subproc import * SOURCE_FILENAME = "corrupt.zip" TMP_DIR = "/tmp" OUTPUT_FILENAME = "rescued.zip" cmdargs = [ "zip", # Linux zip tool "-FF", # or "--fixfix": "fix very broken things" SOURCE_FILENAME, # input file "--temp-path", TMP_DIR, # temporary storage path "--out", OUTPUT_FILENAME # output file ] # We would like to be able to say "y" automatically to # "Is this a single-disk archive? (y/n):" # The source code (api.c, zip.c, zipfile.c), from # ftp://ftp.info-zip.org/pub/infozip/src/ , suggests that "-q" # should do this (internally "-q" sets "noisy = 0") - but in # practice it doesn't work. This is a critical switch. # Therefore we will do something very ugly, and send raw text via # stdin. ZIP_PROMPTS_RESPONSES = [ (SOURCE_STDOUT, "Is this a single-disk archive? (y/n): ", "y\n"), (SOURCE_STDOUT, " or ENTER (try reading this split again): ", "q\n"), (SOURCE_STDERR, "zip: malloc.c:2394: sysmalloc: Assertion `(old_top == initial_top (av) " "&& old_size == 0) || ((unsigned long) (old_size) >= MINSIZE && " "prev_inuse (old_top) && ((unsigned long) old_end & (pagesize - 1)) " "== 0)' failed.", TERMINATE_SUBPROCESS), ] ZIP_STDOUT_TERMINATORS = ["\n", "): "] mimic_user_input(cmdargs, source_challenge_response=ZIP_PROMPTS_RESPONSES, line_terminators=ZIP_STDOUT_TERMINATORS, print_stdout=show_zip_output, print_stdin=show_zip_output)