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 the function will be run again and again, sleeping the stated interval in between.

RepeatedTimers are started, as with threads, by calling their start() method. The timer can be stopped (in between runs) by calling the cancel() 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 the cancel() method. Resetting the counter can be done with the reset() 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 optional stacklevel parameter. If both since and includedoc 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 on b, removing all keys (and associated values) from a that do not appear in b.

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

  • a (dict) – The dictionary to clean against b.

  • b (dict) – The dictionary containing the key structure to clean from a.

Results:

dict: A new dict based on a with all keys (and corresponding values) found in b removed.

octoprint.util.dict_contains_keys(keys, dictionary)

Recursively deep-checks if dictionary contains all keys found in keys.

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
Parameters

  • a (dict) – The dictionary to check for the keys from b.

  • b (dict) – The dictionary whose keys to check a for.

Returns

True if all keys found in b are also present in a, False otherwise.

Return type

boolean

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 and False for values to strip

Returns

A shallow copy of the provided dictionary, stripped of the key-value-pairs

for which the filter_function returned False

Return type

dict

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

  • a (dict) – The dictionary to merge b into

  • b (dict) – The dictionary to merge into a

  • leaf_merger (callable) – An optional callable to use to merge leaves (non-dict values)

  • in_place (boolean) – If set to True, a will be merged with b in place, meaning a will be modified

Returns

b deep-merged into a

Return type

dict

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
Parameters

  • source (dict) – Source dictionary

  • target (dict) – Dictionary to compare to source dictionary and derive diff for

Returns

The minimal dictionary to deep merge on source to get the same result

as deep merging target on source.

Return type

dict

octoprint.util.dict_sanitize(a, b)

Recursively deep-sanitizes a based on b, removing all keys (and associated values) from a that do not appear in b.

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

  • a (dict) – The dictionary to clean against b.

  • b (dict) – The dictionary containing the key structure to clean from a.

Results:

dict: A new dict based on a with all keys (and corresponding values) found in b 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 the extension 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 of gco – iterate through names of the format some_f~1.gco, some_f~2.gco, …, some_~10.gco, some_~11.gco, …, <prefix>~<n>.gco for n 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

type

Raises

ImportError –

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 for extension.

  • 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 provided extension 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 supplied extensions. 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 optional stacklevel parameter. If both since and includedoc 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

bytes

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 optional stacklevel 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 optional stacklevel 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

Parameters

  • command (list, tuple or str) – command to call

  • kwargs (dict) – additional keyword arguments to pass to the sarge run call (note that _async, stdout and stderr will be overwritten)

Returns

(tuple) a 3-tuple of return code, full stdout and full stderr output

checked_call(command, **kwargs)

Calls a command and raises an error if it doesn’t return with return code 0

Parameters

  • command (list, tuple or str) – command to call

  • kwargs (dict) – additional keyword arguments to pass to the sarge run call (note that _async, stdout and stderr will be overwritten)

Returns

(tuple) a 3-tuple of return code, full stdout and full stderr output

Raises

CommandlineError –

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

Parameters

  • returncode (int) – the return code of the command

  • stdout (str) – the stdout output produced by the command

  • stderr (str) – the stderr output produced by the command

class octoprint.util.commandline.DelimiterCapture(delimiter=b'\n', *args, **kwargs)
octoprint.util.commandline.clean_ansi(line)

Removes ANSI control codes from line.

Parameters

line (bytes or str) – the line to process

Returns

(bytes or str) The line without any ANSI control codes

Example:

>>> text = b"Some text with some red words 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) and freebsd (sys.platform = freebsd*).

Returns

(str) mapped OS identifier

octoprint.util.platform.is_os_compatible(compatibility_entries, current_os=None)

Tests if the current_os or sys.platform are blacklisted or whitelisted in compatibility_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.