octoprint.util¶
This module bundles commonly used utility methods or helper classes that are used in multiple places within OctoPrint’s source code.
-
class
octoprint.util.
CaseInsensitiveSet
(*args)¶ Basic case insensitive set
Any str or unicode values will be stored and compared in lower case. Other value types are left as-is.
-
class
octoprint.util.
DefaultOrderedDict
(default_factory=None, *a, **kw)¶ -
copy
() → a shallow copy of od¶
-
-
class
octoprint.util.
PrependableQueue
(maxsize=0)¶
-
class
octoprint.util.
RepeatedTimer
(interval, function, args=None, kwargs=None, run_first=False, condition=None, on_condition_false=None, on_cancelled=None, on_finish=None, daemon=True)¶ This class represents an action that should be run repeatedly in an interval. It is similar to python’s own
threading.Timer
class, but instead of only running once thefunction
will be run again and again, sleeping the statedinterval
in between.RepeatedTimers are started, as with threads, by calling their
start()
method. The timer can be stopped (in between runs) by calling thecancel()
method. The interval the time waited before execution of a loop may not be exactly the same as the interval specified by the user.For example:
def hello(): print("Hello World!") t = RepeatedTimer(1.0, hello) t.start() # prints "Hello World!" every second
Another example with dynamic interval and loop condition:
count = 0 maximum = 5 factor = 1 def interval(): global count global factor return count * factor def condition(): global count global maximum return count <= maximum def hello(): print("Hello World!") global count count += 1 t = RepeatedTimer(interval, hello, run_first=True, condition=condition) t.start() # prints "Hello World!" 5 times, printing the first one # directly, then waiting 1, 2, 3, 4s in between (adaptive interval)
- Parameters
interval (float or callable) – The interval between each
function
call, in seconds. Can also be a callable returning the interval to use, in case the interval is not static.function (callable) – The function to call.
args (list or tuple) – The arguments for the
function
call. Defaults to an empty list.kwargs (dict) – The keyword arguments for the
function
call. Defaults to an empty dict.run_first (boolean) – If set to True, the function will be run for the first time before the first wait period. If set to False (the default), the function will be run for the first time after the first wait period.
condition (callable) – Condition that needs to be True for loop to continue. Defaults to
lambda: True
.on_condition_false (callable) – Callback to call when the timer finishes due to condition becoming false. Will be called before the
on_finish
callback.on_cancelled (callable) – Callback to call when the timer finishes due to being cancelled. Will be called before the
on_finish
callback.on_finish (callable) – Callback to call when the timer finishes, either due to being cancelled or since the condition became false.
daemon (bool) – daemon flag to set on underlying thread.
-
run
()¶ Method representing the thread’s activity.
You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.
-
class
octoprint.util.
ResettableTimer
(interval, function, args=None, kwargs=None, on_reset=None, on_cancelled=None, daemon=True)¶ This class represents an action that should be run after a specified amount of time. It is similar to python’s own
threading.Timer
class, with the addition of being able to reset the counter to zero.ResettableTimers are started, as with threads, by calling their
start()
method. The timer can be stopped (in between runs) by calling thecancel()
method. Resetting the counter can be done with thereset()
method.For example:
def hello(): print("Ran hello() at {}").format(time.time()) t = ResettableTimer(60.0, hello) t.start() print("Started at {}").format(time.time()) time.sleep(30) t.reset() print("Reset at {}").format(time.time())
- Parameters
interval (float or callable) – The interval before calling
function
, in seconds. Can also be a callable returning the interval to use, in case the interval is not static.function (callable) – The function to call.
args (list or tuple) – The arguments for the
function
call. Defaults to an empty list.kwargs (dict) – The keyword arguments for the
function
call. Defaults to an empty dict.on_cancelled (callable) – Callback to call when the timer finishes due to being cancelled.
on_reset (callable) – Callback to call when the timer is reset.
daemon (bool) – daemon flag to set on underlying thread.
-
run
()¶ Method representing the thread’s activity.
You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.
-
exception
octoprint.util.
TypeAlreadyInQueue
(t, *args, **kwargs)¶
-
class
octoprint.util.
TypedQueue
(maxsize=0)¶ -
get
(*args, **kwargs)¶ Remove and return an item from the queue.
If optional args ‘block’ is true and ‘timeout’ is None (the default), block if necessary until an item is available. If ‘timeout’ is a non-negative number, it blocks at most ‘timeout’ seconds and raises the Empty exception if no item was available within that time. Otherwise (‘block’ is false), return an item if one is immediately available, else raise the Empty exception (‘timeout’ is ignored in that case).
-
put
(item, item_type=None, *args, **kwargs)¶ Put an item into the queue.
If optional args ‘block’ is true and ‘timeout’ is None (the default), block if necessary until a free slot is available. If ‘timeout’ is a non-negative number, it blocks at most ‘timeout’ seconds and raises the Full exception if no free slot was available within that time. Otherwise (‘block’ is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (‘timeout’ is ignored in that case).
-
-
octoprint.util.
chunks
(l, n)¶ Yield successive n-sized chunks from l.
Taken from http://stackoverflow.com/a/312464/2028598
-
octoprint.util.
count
(gen)¶ Used instead of len(generator), which doesn’t work
-
octoprint.util.
deprecated
(message, stacklevel=1, since=None, includedoc=None, extenddoc=False)¶ A decorator for deprecated methods. Logs a deprecation warning via Python’s :mod:`warnings module including the supplied
message
. The call stack level used (for adding the source location of the offending call to the warning) can be overridden using the optionalstacklevel
parameter. If bothsince
andincludedoc
are provided, a deprecation warning will also be added to the function’s docstring by providing or extending its__doc__
property.- Parameters
message (string) – The message to include in the deprecation warning.
stacklevel (int) – Stack level for including the caller of the offending method in the logged warning. Defaults to 1, meaning the direct caller of the method. It might make sense to increase this in case of the function call happening dynamically from a fixed position to not shadow the real caller (e.g. in case of overridden
getattr
methods).includedoc (string) – Message about the deprecation to include in the wrapped function’s docstring.
extenddoc (boolean) – If True the original docstring of the wrapped function will be extended by the deprecation message, if False (default) it will be replaced with the deprecation message.
since (string) – Version since when the function was deprecated, must be present for the docstring to get extended.
- Returns
The wrapped function with the deprecation warnings in place.
- Return type
function
-
octoprint.util.
dict_clean
(a, b)¶ Recursively deep-sanitizes
a
based onb
, removing all keys (and associated values) froma
that do not appear inb
.Example:
>>> a = dict(foo="foo", bar="bar", fnord=dict(a=1, b=2, l=["some", "list"])) >>> b = dict(foo=None, fnord=dict(a=None, b=None)) >>> expected = dict(foo="foo", fnord=dict(a=1, b=2)) >>> dict_sanitize(a, b) == expected True >>> dict_clean(a, b) == expected True
- Parameters
- Results:
dict: A new dict based on
a
with all keys (and corresponding values) found inb
removed.
-
octoprint.util.
dict_contains_keys
(keys, dictionary)¶ Recursively deep-checks if
dictionary
contains all keys found inkeys
.Example:
>>> positive = dict(foo="some_other_bar", fnord=dict(b=100)) >>> negative = dict(foo="some_other_bar", fnord=dict(b=100, d=20)) >>> dictionary = dict(foo="bar", fnord=dict(a=1, b=2, c=3)) >>> dict_contains_keys(positive, dictionary) True >>> dict_contains_keys(negative, dictionary) False
-
octoprint.util.
dict_filter
(dictionary, filter_function)¶ Filters a dictionary with the provided filter_function
Example:
>>> data = dict(key1="value1", key2="value2", other_key="other_value", foo="bar", bar="foo") >>> dict_filter(data, lambda k, v: k.startswith("key")) == dict(key1="value1", key2="value2") True >>> dict_filter(data, lambda k, v: v.startswith("value")) == dict(key1="value1", key2="value2") True >>> dict_filter(data, lambda k, v: k == "foo" or v == "foo") == dict(foo="bar", bar="foo") True >>> dict_filter(data, lambda k, v: False) == dict() True >>> dict_filter(data, lambda k, v: True) == data True >>> dict_filter(None, lambda k, v: True) Traceback (most recent call last): ... AssertionError >>> dict_filter(data, None) Traceback (most recent call last): ... AssertionError
- Parameters
dictionary (dict) – The dictionary to filter
filter_function (callable) – The filter function to apply, called with key and value of an entry in the dictionary, must return
True
for values to keep andFalse
for values to strip
- Returns
- A shallow copy of the provided dictionary, stripped of the key-value-pairs
for which the
filter_function
returnedFalse
- Return type
-
octoprint.util.
dict_flatten
(dictionary, prefix='', separator='.')¶ Flatten a dictionary.
- Example::
>>> data = {'a': {'a1': 'a1', 'a2': 'a2'}, 'b': 'b'} >>> expected = {'a.a1': 'a1', 'a.a2': 'a2', 'b': 'b'} >>> actual = dict_flatten(data) >>> shared = {(k, actual[k]) for k in actual if k in expected and actual[k] == expected[k]} >>> len(shared) == len(expected) True
- Parameters
dictionary – the dictionary to flatten
prefix – the key prefix, initially an empty string
separator – key separator, ‘.’ by default
Returns: a flattened dict
-
octoprint.util.
dict_merge
(a, b, leaf_merger=None, in_place=False)¶ Recursively deep-merges two dictionaries.
Based on https://www.xormedia.com/recursively-merge-dictionaries-in-python/
Example:
>>> a = dict(foo="foo", bar="bar", fnord=dict(a=1)) >>> b = dict(foo="other foo", fnord=dict(b=2, l=["some", "list"])) >>> expected = dict(foo="other foo", bar="bar", fnord=dict(a=1, b=2, l=["some", "list"])) >>> dict_merge(a, b) == expected True >>> dict_merge(a, None) == a True >>> dict_merge(None, b) == b True >>> dict_merge(None, None) == dict() True >>> def leaf_merger(a, b): ... if isinstance(a, list) and isinstance(b, list): ... return a + b ... raise ValueError() >>> result = dict_merge(dict(l1=[3, 4], l2=[1], a="a"), dict(l1=[1, 2], l2="foo", b="b"), leaf_merger=leaf_merger) >>> result.get("l1") == [3, 4, 1, 2] True >>> result.get("l2") == "foo" True >>> result.get("a") == "a" True >>> result.get("b") == "b" True >>> c = dict(foo="foo") >>> dict_merge(c, {"bar": "bar"}) is c False >>> dict_merge(c, {"bar": "bar"}, in_place=True) is c True
- Parameters
- Returns
b
deep-merged intoa
- Return type
-
octoprint.util.
dict_minimal_mergediff
(source, target)¶ Recursively calculates the minimal dict that would be needed to be deep merged with a in order to produce the same result as deep merging a and b.
Example:
>>> a = dict(foo=dict(a=1, b=2), bar=dict(c=3, d=4)) >>> b = dict(bar=dict(c=3, d=5), fnord=None) >>> c = dict_minimal_mergediff(a, b) >>> c == dict(bar=dict(d=5), fnord=None) True >>> dict_merge(a, c) == dict_merge(a, b) True
-
octoprint.util.
dict_sanitize
(a, b)¶ Recursively deep-sanitizes
a
based onb
, removing all keys (and associated values) froma
that do not appear inb
.Example:
>>> a = dict(foo="foo", bar="bar", fnord=dict(a=1, b=2, l=["some", "list"])) >>> b = dict(foo=None, fnord=dict(a=None, b=None)) >>> expected = dict(foo="foo", fnord=dict(a=1, b=2)) >>> dict_sanitize(a, b) == expected True >>> dict_clean(a, b) == expected True
- Parameters
- Results:
dict: A new dict based on
a
with all keys (and corresponding values) found inb
removed.
-
class
octoprint.util.
fallback_dict
(custom, *fallbacks)¶ -
items
() → a set-like object providing a view on D’s items¶
-
keys
() → a set-like object providing a view on D’s keys¶
-
values
() → an object providing a view on D’s values¶
-
-
octoprint.util.
filter_non_ascii
(line)¶ Filter predicate to test if a line contains non ASCII characters.
- Parameters
line (string) – The line to test
- Returns
True if the line contains non ASCII characters, False otherwise.
- Return type
boolean
-
octoprint.util.
find_collision_free_name
(filename, extension, existing_filenames, max_power=2)¶ Tries to find a collision free translation of “<filename>.<extension>” to the 8.3 DOS compatible format, preventing collisions with any of the
existing_filenames
.First strips all of
."/\[]:;=,
from the filename and extensions, converts them to lower case and truncates theextension
to a maximum length of 3 characters.If the filename is already equal or less than 8 characters in length after that procedure and “<filename>.<extension>” are not contained in the
existing_files
, that concatenation will be returned as the result.If not, the following algorithm will be applied to try to find a collision free name:
set counter := power := 1 while counter < 10^max_power: set truncated := substr(filename, 0, 6 - power + 1) + "~" + counter set result := "<truncated>.<extension>" if result is collision free: return result counter++ if counter >= 10 ** power: power++ raise ValueError
This will basically – for a given original filename of
some_filename
and an extension ofgco
– iterate through names of the formatsome_f~1.gco
,some_f~2.gco
, …,some_~10.gco
,some_~11.gco
, …,<prefix>~<n>.gco
forn
less than 10 ^max_power
, returning as soon as one is found that is not colliding.- Parameters
filename (string) – The filename without the extension to convert to 8.3.
extension (string) – The extension to convert to 8.3 – will be truncated to 3 characters if it’s longer than that.
existing_filenames (list) – A list of existing filenames to prevent name collisions with.
max_power (int) – Limits the possible attempts of generating a collision free name to 10 ^
max_power
variations. Defaults to 2, so the name generation will maximally reach<name>~99.<ext>
before aborting and raising an exception.
- Returns
- A 8.3 representation of the provided original filename, ensured to not collide with the provided
existing_filenames
- Return type
string
- Raises
ValueError – No collision free name could be found.
Examples
>>> find_collision_free_name("test1234", "gco", []) 'test1234.gco' >>> find_collision_free_name("test1234", "gcode", []) 'test1234.gco' >>> find_collision_free_name("test12345", "gco", []) 'test12~1.gco' >>> find_collision_free_name("test 123", "gco", []) 'test_123.gco' >>> find_collision_free_name("test1234", "g o", []) 'test1234.g_o' >>> find_collision_free_name("test12345", "gco", ["/test12~1.gco"]) 'test12~2.gco' >>> many_files = ["/test12~{}.gco".format(x) for x in range(10)[1:]] >>> find_collision_free_name("test12345", "gco", many_files) 'test1~10.gco' >>> many_more_files = many_files + ["/test1~{}.gco".format(x) for x in range(10, 99)] >>> find_collision_free_name("test12345", "gco", many_more_files) 'test1~99.gco' >>> many_more_files_plus_one = many_more_files + ["/test1~99.gco"] >>> find_collision_free_name("test12345", "gco", many_more_files_plus_one) Traceback (most recent call last): ... ValueError: Can't create a collision free filename >>> find_collision_free_name("test12345", "gco", many_more_files_plus_one, max_power=3) 'test~100.gco'
-
octoprint.util.
get_class
(name)¶ Retrieves the class object for a given fully qualified class name.
- Parameters
name (string) – The fully qualified class name, including all modules separated by
.
- Returns
The class if it could be found.
- Return type
- Raises
-
octoprint.util.
get_dos_filename
(input, existing_filenames=None, extension=None, whitelisted_extensions=None, **kwargs)¶ Converts the provided input filename to a 8.3 DOS compatible filename. If
existing_filenames
is provided, the conversion result will be guaranteed not to collide with any of the filenames provided thus.Uses
find_collision_free_name()
internally.- Parameters
input (string) – The original filename incl. extension to convert to the 8.3 format.
existing_filenames (list) – A list of existing filenames with which the generated 8.3 name must not collide. Optional.
extension (string) – The .3 file extension to use for the generated filename. If not provided, the extension of the provided
filename
will simply be truncated to 3 characters.whitelisted_extensions (list) – A list of extensions on
input
that will be left as-is instead of exchanging forextension
.kwargs (dict) – Additional keyword arguments to provide to
find_collision_free_name()
.
- Returns
- A 8.3 compatible translation of the original filename, not colliding with the optionally provided
existing_filenames
and with the providedextension
or the original extension shortened to a maximum of 3 characters.
- Return type
string
- Raises
ValueError – No 8.3 compatible name could be found that doesn’t collide with the provided
existing_filenames
.
Examples
>>> get_dos_filename("test1234.gco") 'test1234.gco' >>> get_dos_filename("test1234.gcode") 'test1234.gco' >>> get_dos_filename("test12345.gco") 'test12~1.gco' >>> get_dos_filename("test1234.fnord", extension="gco") 'test1234.gco' >>> get_dos_filename("auto0.g", extension="gco") 'auto0.gco' >>> get_dos_filename("auto0.g", extension="gco", whitelisted_extensions=["g"]) 'auto0.g' >>> get_dos_filename(None) >>> get_dos_filename("foo") 'foo'
-
octoprint.util.
get_exception_string
(fmt="{type}: '{message}' @ {file}:{function}:{line}")¶ Retrieves the exception info of the last raised exception and returns it as a string formatted as
<exception type>: <exception message> @ <source file>:<function name>:<line number>
.- Returns
The formatted exception information.
- Return type
string
-
octoprint.util.
get_formatted_datetime
(d)¶ Formats a datetime instance as “YYYY-mm-dd HH:MM” and returns the resulting string.
- Parameters
d (datetime.datetime) – The datetime instance to format
- Returns
The datetime formatted as “YYYY-mm-dd HH:MM”
- Return type
string
-
octoprint.util.
get_formatted_size
(num)¶ Formats the given byte count as a human readable rounded size expressed in the most pressing unit among B(ytes), K(ilo)B(ytes), M(ega)B(ytes), G(iga)B(ytes) and T(era)B(ytes), with one decimal place.
Based on http://stackoverflow.com/a/1094933/2028598
- Parameters
num (int) – The byte count to format
- Returns
The formatted byte count.
- Return type
string
-
octoprint.util.
get_formatted_timedelta
(d)¶ Formats a timedelta instance as “HH:MM:ss” and returns the resulting string.
- Parameters
d (datetime.timedelta) – The timedelta instance to format
- Returns
The timedelta formatted as “HH:MM:ss”
- Return type
string
-
octoprint.util.
get_free_bytes
(path)¶ Deprecated since version 1.2.5: Replaced by psutil.disk_usage.
-
octoprint.util.
get_fully_qualified_classname
(o)¶ Returns the fully qualified class name for an object.
Based on https://stackoverflow.com/a/2020083
- Parameters
o – the object of which to determine the fqcn
- Returns
(str) the fqcn of the object
-
octoprint.util.
is_allowed_file
(filename, extensions)¶ Determines if the provided
filename
has one of the suppliedextensions
. The check is done case-insensitive.- Parameters
filename (string) – The file name to check against the extensions.
extensions (list) – The extensions to check against, a list of strings
- Returns
True if the file name’s extension matches one of the allowed extensions, False otherwise.
- Return type
boolean
-
octoprint.util.
pending_deprecation
(message, stacklevel=1, since=None, includedoc=None, extenddoc=False)¶ A decorator for methods pending deprecation. Logs a pending deprecation warning via Python’s :mod:`warnings module including the supplied
message
. The call stack level used (for adding the source location of the offending call to the warning) can be overridden using the optionalstacklevel
parameter. If bothsince
andincludedoc
are provided, a deprecation warning will also be added to the function’s docstring by providing or extending its__doc__
property.- Parameters
message (string) – The message to include in the deprecation warning.
stacklevel (int) – Stack level for including the caller of the offending method in the logged warning. Defaults to 1, meaning the direct caller of the method. It might make sense to increase this in case of the function call happening dynamically from a fixed position to not shadow the real caller (e.g. in case of overridden
getattr
methods).extenddoc (boolean) – If True the original docstring of the wrapped function will be extended by the deprecation message, if False (default) it will be replaced with the deprecation message.
includedoc (string) – Message about the deprecation to include in the wrapped function’s docstring.
since (string) – Version since when the function was deprecated, must be present for the docstring to get extended.
- Returns
The wrapped function with the deprecation warnings in place.
- Return type
function
-
octoprint.util.
pp
(value)¶ >>> pp(dict()) 'dict()' >>> pp(dict(a=1, b=2, c=3)) 'dict(a=1, b=2, c=3)' >>> pp(set()) 'set()' >>> pp({"a", "b"}) "{'a', 'b'}" >>> pp(["a", "b", "d", "c"]) "['a', 'b', 'd', 'c']" >>> pp(("a", "b", "d", "c")) "('a', 'b', 'd', 'c')" >>> pp("foo") "'foo'" >>> pp([dict(a=1, b=2), {"a", "c", "b"}, (1, 2), None, 1, True, "foo"]) "[dict(a=1, b=2), {'a', 'b', 'c'}, (1, 2), None, 1, True, 'foo']"
-
octoprint.util.
silent_remove
(file)¶ Silently removes a file. Does not raise an error if the file doesn’t exist.
- Parameters
file (string) – The path of the file to be removed
-
octoprint.util.
to_bytes
(s_or_u: Union[str, bytes], encoding: str = 'utf-8', errors: str = 'strict') → Optional[bytes]¶ Make sure
s_or_u
is a byte string.- Parameters
s_or_u (string or unicode) – The value to convert
encoding (string) – encoding to use if necessary, see
str.encode()
errors (string) – error handling to use if necessary, see
str.encode()
- Returns
converted bytes.
- Return type
-
octoprint.util.
to_native_str
(s_or_u: Union[str, bytes], encoding: str = 'utf-8', errors: str = 'strict') → Optional[str]¶ Deprecated since version 2.0.0: to_native_str is no longer needed, use to_unicode
-
octoprint.util.
to_str
(s_or_u: Union[str, bytes], encoding: str = 'utf-8', errors: str = 'strict') → Optional[bytes]¶ Deprecated since version 1.3.11: to_str has been renamed to to_bytes
-
octoprint.util.
to_unicode
(s_or_u: Union[str, bytes], encoding: str = 'utf-8', errors: str = 'strict') → Optional[str]¶ Make sure
s_or_u
is a unicode string.- Parameters
s_or_u (string or unicode) – The value to convert
encoding (string) – encoding to use if necessary, see
bytes.decode()
errors (string) – error handling to use if necessary, see
bytes.decode()
- Returns
converted string.
- Return type
string
-
octoprint.util.
variable_deprecated
(message, stacklevel=1, since=None, includedoc=None, extenddoc=False)¶ A function for deprecated variables. Logs a deprecation warning via Python’s :mod:`warnings module including the supplied
message
. The call stack level used (for adding the source location of the offending call to the warning) can be overridden using the optionalstacklevel
parameter.- Parameters
message (string) – The message to include in the deprecation warning.
stacklevel (int) – Stack level for including the caller of the offending method in the logged warning. Defaults to 1, meaning the direct caller of the method. It might make sense to increase this in case of the function call happening dynamically from a fixed position to not shadow the real caller (e.g. in case of overridden
getattr
methods).since (string) – Version since when the function was deprecated, must be present for the docstring to get extended.
- Returns
The value of the variable with the deprecation warnings in place.
- Return type
value
-
octoprint.util.
variable_pending_deprecation
(message, stacklevel=1, since=None, includedoc=None, extenddoc=False)¶ A decorator for variables pending deprecation. Logs a pending deprecation warning via Python’s :mod:`warnings module including the supplied
message
. The call stack level used (for adding the source location of the offending call to the warning) can be overridden using the optionalstacklevel
parameter.- Parameters
message (string) – The message to include in the deprecation warning.
stacklevel (int) – Stack level for including the caller of the offending method in the logged warning. Defaults to 1, meaning the direct caller of the method. It might make sense to increase this in case of the function call happening dynamically from a fixed position to not shadow the real caller (e.g. in case of overridden
getattr
methods).since (string) – Version since when the function was deprecated, must be present for the docstring to get extended.
- Returns
The value of the variable with the deprecation warnings in place.
- Return type
value
octoprint.util.commandline¶
-
class
octoprint.util.commandline.
CommandlineCaller
¶ The CommandlineCaller is a utility class that allows running command line commands while logging their stdout and stderr via configurable callback functions.
Callbacks are expected to have a signature matching
def callback(*lines): do_something_with_the_passed_lines()
The class utilizes sarge underneath.
Example:
from octoprint.util.commandline import CommandLineCaller, CommandLineError def log(prefix, *lines): for line in lines: print("{} {}".format(prefix, line)) def log_stdout(*lines): log(">>>", *lines) def log_stderr(*lines): log("!!!", *lines) def log_call(*lines) log("---", *lines) caller = CommandLineCaller() caller.on_log_call = log_call caller.on_log_stdout = log_stdout caller.on_log_stderr = log_stderr try: caller.checked_call(["some", "command", "with", "parameters"]) except CommandLineError as err: print("Command returned {}".format(err.returncode)) else: print("Command finished successfully")
-
call
(command, **kwargs)¶ Calls a command
-
checked_call
(command, **kwargs)¶ Calls a command and raises an error if it doesn’t return with return code 0
- Parameters
- Returns
(tuple) a 3-tuple of return code, full stdout and full stderr output
- Raises
-
on_log_call
¶ Callback for the called command line
-
on_log_stderr
¶ Callback for stderr output
-
on_log_stdout
¶ Callback for stdout output
-
-
exception
octoprint.util.commandline.
CommandlineError
(returncode, stdout, stderr)¶ Raised by
checked_call()
on non zero return codes
-
class
octoprint.util.commandline.
DelimiterCapture
(delimiter=b'\n', *args, **kwargs)¶
-
octoprint.util.commandline.
clean_ansi
(line)¶ Removes ANSI control codes from
line
.- Parameters
- Returns
(bytes or str) The line without any ANSI control codes
Example:
>>> text = b"Some text with some [31mred words[39m in it" >>> clean_ansi(text) 'Some text with some red words in it' >>> text = b"We [?25lhide the cursor here and then [?25hshow it again here" >>> clean_ansi(text) 'We hide the cursor here and then show it again here'
octoprint.util.platform¶
This module bundles platform specific flags and implementations.
-
octoprint.util.platform.
CLOSE_FDS
= True¶ Default setting for close_fds parameter to Popen/sarge.run.
Set
close_fds
on every sub process to this to ensure file handlers will be closed on child processes on platforms that support this (anything Python 3.7+ or anything but win32 in earlier Python versions).
-
octoprint.util.platform.
get_os
()¶ Returns a canonical OS identifier.
Currently the following OS are recognized:
win32
,linux
(sys.platform
=linux*
),macos
(sys.platform
=darwin
) andfreebsd
(sys.platform
=freebsd*
).- Returns
(str) mapped OS identifier
-
octoprint.util.platform.
is_os_compatible
(compatibility_entries, current_os=None)¶ Tests if the
current_os
orsys.platform
are blacklisted or whitelisted incompatibility_entries
- Returns
(bool) True if the os is compatible, False otherwise
-
octoprint.util.platform.
set_close_exec
(handle)¶ Set
close_exec
flag on handle, if supported by the OS.