kiwi.command_process

View Source

  1# Copyright (c) 2015 SUSE Linux GmbH.  All rights reserved.
  2#
  3# This file is part of kiwi.
  4#
  5# kiwi is free software: you can redistribute it and/or modify
  6# it under the terms of the GNU General Public License as published by
  7# the Free Software Foundation, either version 3 of the License, or
  8# (at your option) any later version.
  9#
 10# kiwi is distributed in the hope that it will be useful,
 11# but WITHOUT ANY WARRANTY; without even the implied warranty of
 12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 13# GNU General Public License for more details.
 14#
 15# You should have received a copy of the GNU General Public License
 16# along with kiwi.  If not, see <http://www.gnu.org/licenses/>
 17#
 18import os
 19import logging
 20from collections import namedtuple
 21from kiwi.command import CommandCallT
 22from typing import (
 23    NamedTuple, List, Callable
 24)
 25
 26# project
 27from kiwi.utils.codec import Codec
 28from kiwi.logger import Logger
 29
 30from kiwi.exceptions import KiwiCommandError
 31
 32log = logging.getLogger('kiwi')
 33
 34
 35class PollT(NamedTuple):
 36    stdout_line: str
 37    stderr_line: str
 38
 39
 40class CommandProcess:
 41    """
 42    **Implements processing of non blocking Command calls**
 43
 44    Provides methods to iterate over non blocking instances of
 45    the Command class with and without progress information
 46
 47    :param subprocess command: instance of subprocess
 48    :param string log_topic: topic string for logging
 49    """
 50    def __init__(self, command: CommandCallT, log_topic='system') -> None:
 51        self.command = CommandIterator(command)
 52        self.log_topic = log_topic
 53        self.items_processed = 0
 54
 55    def poll(self):
 56        """
 57        Iterate over process, raise on error and log output
 58        """
 59        for lineT in self.command:
 60            line = lineT.stdout_line
 61            if line:
 62                log.debug('%s: %s', self.log_topic, line)
 63        if self.command.get_error_code() != 0:
 64            raise KiwiCommandError(
 65                self.command.get_error_output()
 66            )
 67
 68    def poll_show_progress(
 69        self, items_to_complete: List[str], match_method: Callable,
 70        with_stderr: bool = False
 71    ):
 72        """
 73        Iterate over process and show progress in percent
 74        raise on error and log output
 75
 76        :param list items_to_complete: all items
 77        :param function match_method: method matching item
 78        """
 79        self._init_progress()
 80        for lineT in self.command:
 81            lines = [lineT.stdout_line]
 82            if with_stderr:
 83                lines.append(lineT.stderr_line)
 84            for line in lines:
 85                if line:
 86                    log.debug('%s: %s', self.log_topic, line)
 87                    self._update_progress(
 88                        match_method, items_to_complete, line
 89                    )
 90        self._stop_progress()
 91        if self.command.get_error_code() != 0:
 92            raise KiwiCommandError(
 93                self.command.get_error_output()
 94            )
 95
 96    def poll_and_watch(self):
 97        """
 98        Iterate over process don't raise on error and log
 99        stdout and stderr
100        """
101        log.info(self.log_topic)
102        log.debug('--------------out start-------------')
103        for lineT in self.command:
104            line = lineT.stdout_line
105            if line:
106                log.debug(line)
107        log.debug('--------------out stop--------------')
108
109        error_code = self.command.get_error_code()
110        error_output = self.command.get_error_output()
111        result = namedtuple(
112            'result', ['stderr', 'returncode']
113        )
114        if error_output:
115            log.debug('--------------err start-------------')
116            for line in error_output.split(os.linesep):
117                log.debug(line)
118            log.debug('--------------err stop--------------')
119        return result(
120            stderr=error_output, returncode=error_code
121        )
122
123    def create_match_method(self, method):
124        """
125        create a matcher function pointer which calls the given
126        method as method(item_to_match, data) on dereference
127
128        :param function method: function reference
129
130        :return: function pointer
131        :rtype: object
132        """
133        def create_method(item_to_match, data):
134            return method(item_to_match, data)
135        return create_method
136
137    def returncode(self):
138        return self.command.get_error_code()
139
140    def _init_progress(self):
141        Logger.progress(
142            0, 100, '[ INFO    ]: Processing'
143        )
144
145    def _stop_progress(self):
146        Logger.progress(
147            100, 100, '[ INFO    ]: Processing'
148        )
149
150    def _update_progress(
151        self, match_method, items_to_complete, command_output
152    ):
153        items_count = len(items_to_complete)
154        for item in items_to_complete:
155            if match_method(item, command_output):
156                self.items_processed += 1
157                if self.items_processed <= items_count:
158                    Logger.progress(
159                        self.items_processed, items_count,
160                        '[ INFO    ]: Processing'
161                    )
162
163
164class CommandIterator:
165    """
166    **Implements an Iterator for Instances of Command**
167
168    :param subprocess command: instance of subprocess
169    """
170    def __init__(self, command: CommandCallT) -> None:
171        self.command = command
172        self.command_error_output = bytes(b'')
173        self.command_output_line = bytes(b'')
174        self.command_error_line = bytes(b'')
175        self.output_eof_reached = False
176        self.errors_eof_reached = False
177
178    def __next__(self) -> PollT:
179        line_stdout = ''
180        line_stderr = ''
181        if self.command.process.poll() is not None:
182            if self.output_eof_reached and self.errors_eof_reached:
183                raise StopIteration()
184
185        if self.command.output_available():
186            byte_read = self.command.output.read(1)
187            if not byte_read:
188                self.output_eof_reached = True
189            elif byte_read == bytes(b'\n'):
190                line_stdout = Codec.decode(self.command_output_line)
191                self.command_output_line = bytes(b'')
192            else:
193                self.command_output_line += byte_read
194
195        if self.command.error_available():
196            byte_read = self.command.error.read(1)
197            if not byte_read:
198                self.errors_eof_reached = True
199            elif byte_read == bytes(b'\n'):
200                line_stderr = Codec.decode(self.command_error_line)
201                self.command_error_line = bytes(b'')
202                self.command_error_output += byte_read
203            else:
204                self.command_error_line += byte_read
205                self.command_error_output += byte_read
206
207        return PollT(
208            stdout_line=line_stdout,
209            stderr_line=line_stderr
210        )
211
212    def get_error_output(self):
213        """
214        Provide data which was sent to the stderr channel
215
216        :return: stderr data
217
218        :rtype: str
219        """
220        return Codec.decode(self.command_error_output)
221
222    def get_error_code(self) -> int:
223        """
224        Provide return value from processed command
225
226        :return: errorcode
227
228        :rtype: int
229        """
230        return self.command.process.returncode
231
232    def get_pid(self) -> int:
233        """
234        Provide process ID of command while running
235
236        :return: pid
237
238        :rtype: int
239        """
240        return self.command.process.pid
241
242    def kill(self) -> None:
243        """
244        Send kill signal SIGTERM to command process
245        """
246        self.command.process.kill()
247
248    def __iter__(self):
249        return self

log = <Logger kiwi (DEBUG)>

class PollT(typing.NamedTuple): View Source

36class PollT(NamedTuple):
37    stdout_line: str
38    stderr_line: str

PollT(stdout_line, stderr_line)

PollT(stdout_line: str, stderr_line: str)

Create new instance of PollT(stdout_line, stderr_line)

stdout_line: str

Alias for field number 0

stderr_line: str

Alias for field number 1

class CommandProcess: View Source

 41class CommandProcess:
 42    """
 43    **Implements processing of non blocking Command calls**
 44
 45    Provides methods to iterate over non blocking instances of
 46    the Command class with and without progress information
 47
 48    :param subprocess command: instance of subprocess
 49    :param string log_topic: topic string for logging
 50    """
 51    def __init__(self, command: CommandCallT, log_topic='system') -> None:
 52        self.command = CommandIterator(command)
 53        self.log_topic = log_topic
 54        self.items_processed = 0
 55
 56    def poll(self):
 57        """
 58        Iterate over process, raise on error and log output
 59        """
 60        for lineT in self.command:
 61            line = lineT.stdout_line
 62            if line:
 63                log.debug('%s: %s', self.log_topic, line)
 64        if self.command.get_error_code() != 0:
 65            raise KiwiCommandError(
 66                self.command.get_error_output()
 67            )
 68
 69    def poll_show_progress(
 70        self, items_to_complete: List[str], match_method: Callable,
 71        with_stderr: bool = False
 72    ):
 73        """
 74        Iterate over process and show progress in percent
 75        raise on error and log output
 76
 77        :param list items_to_complete: all items
 78        :param function match_method: method matching item
 79        """
 80        self._init_progress()
 81        for lineT in self.command:
 82            lines = [lineT.stdout_line]
 83            if with_stderr:
 84                lines.append(lineT.stderr_line)
 85            for line in lines:
 86                if line:
 87                    log.debug('%s: %s', self.log_topic, line)
 88                    self._update_progress(
 89                        match_method, items_to_complete, line
 90                    )
 91        self._stop_progress()
 92        if self.command.get_error_code() != 0:
 93            raise KiwiCommandError(
 94                self.command.get_error_output()
 95            )
 96
 97    def poll_and_watch(self):
 98        """
 99        Iterate over process don't raise on error and log
100        stdout and stderr
101        """
102        log.info(self.log_topic)
103        log.debug('--------------out start-------------')
104        for lineT in self.command:
105            line = lineT.stdout_line
106            if line:
107                log.debug(line)
108        log.debug('--------------out stop--------------')
109
110        error_code = self.command.get_error_code()
111        error_output = self.command.get_error_output()
112        result = namedtuple(
113            'result', ['stderr', 'returncode']
114        )
115        if error_output:
116            log.debug('--------------err start-------------')
117            for line in error_output.split(os.linesep):
118                log.debug(line)
119            log.debug('--------------err stop--------------')
120        return result(
121            stderr=error_output, returncode=error_code
122        )
123
124    def create_match_method(self, method):
125        """
126        create a matcher function pointer which calls the given
127        method as method(item_to_match, data) on dereference
128
129        :param function method: function reference
130
131        :return: function pointer
132        :rtype: object
133        """
134        def create_method(item_to_match, data):
135            return method(item_to_match, data)
136        return create_method
137
138    def returncode(self):
139        return self.command.get_error_code()
140
141    def _init_progress(self):
142        Logger.progress(
143            0, 100, '[ INFO    ]: Processing'
144        )
145
146    def _stop_progress(self):
147        Logger.progress(
148            100, 100, '[ INFO    ]: Processing'
149        )
150
151    def _update_progress(
152        self, match_method, items_to_complete, command_output
153    ):
154        items_count = len(items_to_complete)
155        for item in items_to_complete:
156            if match_method(item, command_output):
157                self.items_processed += 1
158                if self.items_processed <= items_count:
159                    Logger.progress(
160                        self.items_processed, items_count,
161                        '[ INFO    ]: Processing'
162                    )

Implements processing of non blocking Command calls

Provides methods to iterate over non blocking instances of the Command class with and without progress information

Parameters

subprocess command: instance of subprocess
string log_topic: topic string for logging

CommandProcess(command: kiwi.command.CommandCallT, log_topic='system') View Source

51    def __init__(self, command: CommandCallT, log_topic='system') -> None:
52        self.command = CommandIterator(command)
53        self.log_topic = log_topic
54        self.items_processed = 0

command

log_topic

items_processed

def poll(self): View Source

56    def poll(self):
57        """
58        Iterate over process, raise on error and log output
59        """
60        for lineT in self.command:
61            line = lineT.stdout_line
62            if line:
63                log.debug('%s: %s', self.log_topic, line)
64        if self.command.get_error_code() != 0:
65            raise KiwiCommandError(
66                self.command.get_error_output()
67            )

Iterate over process, raise on error and log output

def poll_show_progress( self, items_to_complete: List[str], match_method: Callable, with_stderr: bool = False): View Source

69    def poll_show_progress(
70        self, items_to_complete: List[str], match_method: Callable,
71        with_stderr: bool = False
72    ):
73        """
74        Iterate over process and show progress in percent
75        raise on error and log output
76
77        :param list items_to_complete: all items
78        :param function match_method: method matching item
79        """
80        self._init_progress()
81        for lineT in self.command:
82            lines = [lineT.stdout_line]
83            if with_stderr:
84                lines.append(lineT.stderr_line)
85            for line in lines:
86                if line:
87                    log.debug('%s: %s', self.log_topic, line)
88                    self._update_progress(
89                        match_method, items_to_complete, line
90                    )
91        self._stop_progress()
92        if self.command.get_error_code() != 0:
93            raise KiwiCommandError(
94                self.command.get_error_output()
95            )

Iterate over process and show progress in percent raise on error and log output

Parameters

list items_to_complete: all items
function match_method: method matching item

def poll_and_watch(self): View Source

 97    def poll_and_watch(self):
 98        """
 99        Iterate over process don't raise on error and log
100        stdout and stderr
101        """
102        log.info(self.log_topic)
103        log.debug('--------------out start-------------')
104        for lineT in self.command:
105            line = lineT.stdout_line
106            if line:
107                log.debug(line)
108        log.debug('--------------out stop--------------')
109
110        error_code = self.command.get_error_code()
111        error_output = self.command.get_error_output()
112        result = namedtuple(
113            'result', ['stderr', 'returncode']
114        )
115        if error_output:
116            log.debug('--------------err start-------------')
117            for line in error_output.split(os.linesep):
118                log.debug(line)
119            log.debug('--------------err stop--------------')
120        return result(
121            stderr=error_output, returncode=error_code
122        )

Iterate over process don't raise on error and log stdout and stderr

def create_match_method(self, method): View Source

124    def create_match_method(self, method):
125        """
126        create a matcher function pointer which calls the given
127        method as method(item_to_match, data) on dereference
128
129        :param function method: function reference
130
131        :return: function pointer
132        :rtype: object
133        """
134        def create_method(item_to_match, data):
135            return method(item_to_match, data)
136        return create_method

create a matcher function pointer which calls the given method as method(item_to_match, data) on dereference

Parameters

function method: function reference

Returns

function pointer

def returncode(self): View Source

138    def returncode(self):
139        return self.command.get_error_code()

class CommandIterator: View Source

165class CommandIterator:
166    """
167    **Implements an Iterator for Instances of Command**
168
169    :param subprocess command: instance of subprocess
170    """
171    def __init__(self, command: CommandCallT) -> None:
172        self.command = command
173        self.command_error_output = bytes(b'')
174        self.command_output_line = bytes(b'')
175        self.command_error_line = bytes(b'')
176        self.output_eof_reached = False
177        self.errors_eof_reached = False
178
179    def __next__(self) -> PollT:
180        line_stdout = ''
181        line_stderr = ''
182        if self.command.process.poll() is not None:
183            if self.output_eof_reached and self.errors_eof_reached:
184                raise StopIteration()
185
186        if self.command.output_available():
187            byte_read = self.command.output.read(1)
188            if not byte_read:
189                self.output_eof_reached = True
190            elif byte_read == bytes(b'\n'):
191                line_stdout = Codec.decode(self.command_output_line)
192                self.command_output_line = bytes(b'')
193            else:
194                self.command_output_line += byte_read
195
196        if self.command.error_available():
197            byte_read = self.command.error.read(1)
198            if not byte_read:
199                self.errors_eof_reached = True
200            elif byte_read == bytes(b'\n'):
201                line_stderr = Codec.decode(self.command_error_line)
202                self.command_error_line = bytes(b'')
203                self.command_error_output += byte_read
204            else:
205                self.command_error_line += byte_read
206                self.command_error_output += byte_read
207
208        return PollT(
209            stdout_line=line_stdout,
210            stderr_line=line_stderr
211        )
212
213    def get_error_output(self):
214        """
215        Provide data which was sent to the stderr channel
216
217        :return: stderr data
218
219        :rtype: str
220        """
221        return Codec.decode(self.command_error_output)
222
223    def get_error_code(self) -> int:
224        """
225        Provide return value from processed command
226
227        :return: errorcode
228
229        :rtype: int
230        """
231        return self.command.process.returncode
232
233    def get_pid(self) -> int:
234        """
235        Provide process ID of command while running
236
237        :return: pid
238
239        :rtype: int
240        """
241        return self.command.process.pid
242
243    def kill(self) -> None:
244        """
245        Send kill signal SIGTERM to command process
246        """
247        self.command.process.kill()
248
249    def __iter__(self):
250        return self

Implements an Iterator for Instances of Command

Parameters

subprocess command: instance of subprocess

CommandIterator(command: kiwi.command.CommandCallT) View Source

171    def __init__(self, command: CommandCallT) -> None:
172        self.command = command
173        self.command_error_output = bytes(b'')
174        self.command_output_line = bytes(b'')
175        self.command_error_line = bytes(b'')
176        self.output_eof_reached = False
177        self.errors_eof_reached = False

command

command_error_output

command_output_line

command_error_line

output_eof_reached

errors_eof_reached

def get_error_output(self): View Source

213    def get_error_output(self):
214        """
215        Provide data which was sent to the stderr channel
216
217        :return: stderr data
218
219        :rtype: str
220        """
221        return Codec.decode(self.command_error_output)

Provide data which was sent to the stderr channel

Returns

stderr data

def get_error_code(self) -> int: View Source

223    def get_error_code(self) -> int:
224        """
225        Provide return value from processed command
226
227        :return: errorcode
228
229        :rtype: int
230        """
231        return self.command.process.returncode

Provide return value from processed command

Returns

errorcode

def get_pid(self) -> int: View Source

233    def get_pid(self) -> int:
234        """
235        Provide process ID of command while running
236
237        :return: pid
238
239        :rtype: int
240        """
241        return self.command.process.pid

Provide process ID of command while running

Returns

pid

def kill(self) -> None: View Source

243    def kill(self) -> None:
244        """
245        Send kill signal SIGTERM to command process
246        """
247        self.command.process.kill()

Send kill signal SIGTERM to command process