0
# Utility Functions
1
2
Helper functions, constants, and utilities that support the core command_runner functionality. Includes encoding conversion, threading support, and platform-specific constants.
3
4
## Capabilities
5
6
### Encoding and Output Conversion
7
8
Robust text encoding conversion with comprehensive error handling for different platforms and character sets.
9
10
```python { .api }
11
def to_encoding(process_output, encoding, errors):
12
"""
13
Convert bytes output to string with comprehensive error handling.
14
15
Handles the conversion of subprocess output from bytes to strings with
16
platform-appropriate encoding and error handling strategies. Supports
17
various encoding scenarios and provides fallback mechanisms.
18
19
Args:
20
process_output (Union[str, bytes]): Raw process output to convert
21
encoding (Optional[str]): Target encoding string. None bypasses conversion.
22
False returns original bytes unchanged.
23
errors (str): Error handling strategy for encoding issues:
24
'backslashreplace', 'ignore', 'replace', 'strict'
25
26
Returns:
27
str: Converted string output, or original if conversion not needed
28
29
Examples:
30
Convert Windows command output:
31
>>> output = b'Hello\xff World'
32
>>> result = to_encoding(output, 'cp437', 'backslashreplace')
33
>>> print(result) # 'Hello\\xff World'
34
35
Handle UTF-8 conversion:
36
>>> output = b'Caf\xc3\xa9'
37
>>> result = to_encoding(output, 'utf-8', 'strict')
38
>>> print(result) # 'Café'
39
40
Skip encoding conversion:
41
>>> output = b'binary data'
42
>>> result = to_encoding(output, False, 'ignore')
43
>>> print(type(result)) # <class 'bytes'>
44
45
Error Handling:
46
- Gracefully handles encoding errors with specified strategy
47
- Converts None output to empty string
48
- Logs encoding errors for debugging
49
- Falls back to 'ignore' mode for problematic encodings
50
51
Platform Notes:
52
Windows: Commonly used with 'cp437' encoding for cmd.exe output
53
Unix/Linux: Typically used with 'utf-8' encoding
54
macOS: Usually 'utf-8' encoding with some locale variations
55
"""
56
```
57
58
### Threading Support
59
60
Decorator and helper functions for asynchronous command execution using Python's concurrent.futures.
61
62
```python { .api }
63
def threaded(fn):
64
"""
65
Decorator to make any function return a concurrent.Future object.
66
67
Converts synchronous functions into asynchronous versions that execute
68
in background threads and return Future objects for result handling.
69
70
Args:
71
fn: Function to make threaded
72
73
Returns:
74
Wrapped function that returns concurrent.futures.Future
75
76
Examples:
77
Create threaded function:
78
>>> @threaded
79
... def long_task(duration):
80
... time.sleep(duration)
81
... return f"Completed after {duration} seconds"
82
>>>
83
>>> future = long_task(5)
84
>>> print("Task started...")
85
>>> result = future.result() # Blocks until complete
86
>>> print(result)
87
88
Force synchronous execution:
89
>>> future = long_task(5, __no_threads=True) # Executes immediately
90
91
Implementation:
92
Uses concurrent.futures.Future with background thread execution
93
Available only on Python 3.3+ (graceful fallback on older versions)
94
Supports special __no_threads keyword to force synchronous execution
95
"""
96
97
def call_with_future(fn, future, args, kwargs):
98
"""
99
Execute function with Future result handling.
100
101
Helper function for threaded execution that properly handles function
102
results and exceptions in the Future object.
103
104
Args:
105
fn: Function to execute
106
future: Future object to store result/exception
107
args: Positional arguments for function
108
kwargs: Keyword arguments for function
109
110
Returns:
111
None (result stored in future object)
112
113
Implementation:
114
Called internally by @threaded decorator
115
Handles both successful results and exceptions
116
Ensures proper Future state management
117
"""
118
```
119
120
### Deferred Execution
121
122
Launch detached processes with time delays for cleanup and maintenance operations.
123
124
```python { .api }
125
def deferred_command(command, defer_time=300):
126
"""
127
Launch a command detached from parent process after specified delay.
128
129
Creates an independent shell process that waits for the specified time
130
then executes the command. Useful for self-updating applications, cleanup
131
operations, or any task that needs to run after the parent process exits.
132
133
Args:
134
command (str): Shell command to execute after delay
135
defer_time (int): Delay in seconds before execution (default: 300)
136
137
Returns:
138
None
139
140
Examples:
141
Cleanup temporary files after 5 minutes:
142
>>> deferred_command('rm -rf /tmp/myapp_temp', defer_time=300)
143
144
Self-update after application exit:
145
>>> deferred_command('cp /tmp/newversion /usr/local/bin/myapp', defer_time=10)
146
147
Log rotation:
148
>>> deferred_command('gzip /var/log/myapp.log && mv /var/log/myapp.log.gz /archive/', defer_time=3600)
149
150
Platform Implementation:
151
Windows: Uses 'ping 127.0.0.1 -n {seconds}' as timer, then executes command
152
Unix/Linux: Uses 'sleep {seconds} && command' pattern
153
154
Process Details:
155
Creates completely detached subprocess with no stdio connections
156
Parent process can exit safely without affecting deferred command
157
Command executes in shell context for maximum flexibility
158
No output capture or error handling (fire-and-forget operation)
159
"""
160
```
161
162
## Constants and Enumerations
163
164
### Subprocess Constants
165
166
```python { .api }
167
# Standard subprocess pipe reference
168
PIPE = subprocess.PIPE
169
170
# Platform-specific priority constants (Windows)
171
BELOW_NORMAL_PRIORITY_CLASS = 16384
172
HIGH_PRIORITY_CLASS = 128
173
NORMAL_PRIORITY_CLASS = 32
174
REALTIME_PRIORITY_CLASS = 256
175
IDLE_PRIORITY_CLASS = 64
176
177
# IO Priority constants (Windows)
178
IOPRIO_HIGH = 3
179
IOPRIO_NORMAL = 2
180
IOPRIO_LOW = 1
181
182
# IO Priority constants (Unix/Linux)
183
IOPRIO_CLASS_IDLE = 3
184
IOPRIO_CLASS_BE = 2 # Best Effort
185
IOPRIO_CLASS_RT = 1 # Real Time
186
```
187
188
### Priority Mapping Dictionary
189
190
```python { .api }
191
PRIORITIES = {
192
"process": {
193
"verylow": ..., # Platform-specific idle priority
194
"low": ..., # Below normal priority
195
"normal": ..., # Standard priority
196
"high": ..., # Above normal priority
197
"rt": ... # Real-time priority
198
},
199
"io": {
200
"low": ..., # Background IO priority
201
"normal": ..., # Standard IO priority
202
"high": ... # High-priority IO
203
}
204
}
205
```
206
207
## Exception Classes
208
209
Custom exception hierarchy for specialized error handling and output preservation.
210
211
```python { .api }
212
class TimeoutExpired(BaseException):
213
"""
214
Command timeout exception with output preservation.
215
216
Compatible backport of subprocess.TimeoutExpired for Python <= 3.3.
217
Preserves partial command output when timeout occurs.
218
"""
219
def __init__(self, cmd, timeout, output=None, stderr=None):
220
"""
221
Initialize timeout exception.
222
223
Args:
224
cmd: Command that timed out
225
timeout: Timeout value in seconds
226
output: Partial stdout output
227
stderr: Partial stderr output
228
"""
229
230
@property
231
def stdout(self):
232
"""Alias for output property for compatibility."""
233
234
class InterruptGetOutput(BaseException):
235
"""
236
Base exception for capturing output during interruptions.
237
238
Preserves command output when execution is interrupted by various
239
conditions (timeouts, stop conditions, keyboard interrupts).
240
"""
241
def __init__(self, output):
242
"""
243
Initialize with preserved output.
244
245
Args:
246
output: Partial command output to preserve
247
"""
248
249
@property
250
def output(self):
251
"""Access preserved output from interrupted command."""
252
253
class KbdInterruptGetOutput(InterruptGetOutput):
254
"""
255
Keyboard interrupt with output preservation.
256
257
Raised when KeyboardInterrupt (Ctrl+C) occurs during command execution.
258
Preserves any output captured before interruption.
259
"""
260
261
class StopOnInterrupt(InterruptGetOutput):
262
"""
263
Stop condition interrupt with output preservation.
264
265
Raised when custom stop_on function returns True during execution.
266
Preserves any output captured before stop condition triggered.
267
"""
268
```
269
270
## Module-Level Variables
271
272
Version and metadata information for the command_runner package.
273
274
```python { .api }
275
# Package metadata
276
__version__ = "1.7.4"
277
__author__ = "Orsiris de Jong"
278
__copyright__ = "Copyright (C) 2015-2025 Orsiris de Jong for NetInvent"
279
__licence__ = "BSD 3 Clause"
280
__build__ = "2025052301"
281
__compat__ = "python2.7+"
282
283
# Platform detection
284
os_name = os.name # 'nt' for Windows, 'posix' for Unix/Linux
285
286
# Logger instance
287
logger = getLogger(__intname__)
288
```
289
290
## Platform Compatibility Utilities
291
292
### Python Version Compatibility
293
294
Command Runner includes compatibility shims for older Python versions:
295
296
```python
297
# Python 2.7 compatibility imports
298
try:
299
import queue
300
except ImportError:
301
import Queue as queue
302
303
try:
304
from typing import Union, Optional, List, Tuple, Any, Callable
305
except ImportError:
306
pass # Type hints not available in Python 2.7
307
308
# Python <= 3.3 compatibility
309
try:
310
TimeoutExpired = subprocess.TimeoutExpired
311
except AttributeError:
312
# Custom TimeoutExpired class for older Python versions
313
class TimeoutExpired(BaseException): ...
314
```
315
316
### Dependency Detection
317
318
Graceful handling of optional dependencies:
319
320
```python
321
# Optional psutil import with fallback
322
try:
323
import psutil
324
HAS_PSUTIL = True
325
except ImportError:
326
HAS_PSUTIL = False
327
# Fallback priority constants defined
328
329
# Optional signal module handling
330
try:
331
import signal
332
HAS_SIGNAL = True
333
except ImportError:
334
HAS_SIGNAL = False
335
```
336
337
## Usage Patterns
338
339
### Custom Encoding Scenarios
340
341
```python
342
from command_runner import command_runner, to_encoding
343
344
# Windows PowerShell with Unicode output
345
exit_code, raw_output = command_runner(
346
'powershell -Command "Get-Process"',
347
encoding=False # Get raw bytes
348
)
349
converted = to_encoding(raw_output, 'unicode_escape', 'backslashreplace')
350
351
# Linux with mixed encoding
352
exit_code, output = command_runner(
353
'ls -la /path/with/mixed/encoding',
354
encoding='utf-8',
355
errors='replace' # Replace invalid characters
356
)
357
```
358
359
### Threading Integration
360
361
```python
362
from command_runner import command_runner_threaded
363
import queue
364
import threading
365
366
def process_command_output():
367
"""Example of threaded command with live output processing."""
368
output_queue = queue.Queue()
369
370
# Start command in background
371
future = command_runner_threaded(
372
'long_running_command',
373
stdout=output_queue,
374
method='poller'
375
)
376
377
# Process output as it arrives
378
while not future.done():
379
try:
380
line = output_queue.get(timeout=0.1)
381
if line is None:
382
break
383
process_line(line)
384
except queue.Empty:
385
continue
386
387
# Get final result
388
exit_code, output = future.result()
389
return exit_code, output
390
```
391
392
### Error Handling with Preserved Output
393
394
```python
395
from command_runner import command_runner, TimeoutExpired, KbdInterruptGetOutput
396
397
try:
398
exit_code, output = command_runner('long_command', timeout=30)
399
except TimeoutExpired as e:
400
print(f"Command timed out after {e.timeout} seconds")
401
print(f"Partial output: {e.output}")
402
except KbdInterruptGetOutput as e:
403
print("Command interrupted by user")
404
print(f"Partial output: {e.output}")
405
```