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

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.deserialize(filename, encoding='utf-8')¶

Deserializes data from a file

In the current implementation this uses json.loads and - if the file is found to be compressed - zlib.decompress.

Parameters:¶

filename : str¶: The file to deserialize from
encoding : str¶: The encoding to use for the file, defaults to utf-8

Returns:¶

The deserialized data structure

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_all_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_all_keys(positive, dictionary)
True
>>> dict_contains_all_keys(negative, dictionary)
False

Parameters:¶

keys : dict¶: The dictionary whose keys to check dictionary for.
dictionary : dict¶: The dictionary to check for the keys from keys.

Returns:¶

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

Return type:¶

boolean

octoprint.util.dict_contains_any_keys(keys, dictionary)¶

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

Example:

>>> positive = dict(foo="some_other_bar")
>>> negative = dict(fnord=dict(e=True))
>>> dictionary = dict(foo="bar", fnord=dict(a=1, b=2, c=3))
>>> dict_contains_any_keys(positive, dictionary)
True
>>> dict_contains_any_keys(negative, dictionary)
False

Parameters:¶

keys : dict¶: The dictionary whose keys to check dictionary for.
dictionary : dict¶: The dictionary to check for the keys from keys.

Returns:¶

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

Return type:¶

boolean

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_all_keys(positive, dictionary)
True
>>> dict_contains_all_keys(negative, dictionary)
False

Parameters:¶

keys : dict¶: The dictionary whose keys to check dictionary for.
dictionary : dict¶: The dictionary to check for the keys from keys.

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.get_bom(filename, encoding)¶

Check if the file has a BOM and if so return it.

Params:: filename (str): The file to check. encoding (str): The encoding to check for.

Returns:¶: (bytes) the BOM or None if there is no BOM.

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_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_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.serialize(filename, data, encoding='utf-8', compressed=True)¶

Serializes data to a file

In the current implementation this uses json.dumps.

If compressed is True (the default), the serialized data put through zlib.compress.

Supported data types are listed at the bottom of octoprint.util.comprehensive_json(), and include some data types that are not supported by json.dumps by default.

This is not thread-safe, if concurrent access is required, the caller needs to ensure that only one thread is writing to the file at any given time.

Parameters:¶

filename : str¶: The file to write to
data : object¶: The data to serialize
encoding : str¶: The encoding to use for the file
compressed : bool¶: Whether to compress the data before writing it to the file

octoprint.util.to_bytes(s_or_u: str | bytes, encoding: str = 'utf-8', errors: str = 'strict') → bytes¶

Make sure s_or_u is a byte string.

Parameters:¶

s_or_u : str or bytes¶: The value to convert
encoding : str¶: encoding to use if necessary, see str.encode()
errors : str¶: error handling to use if necessary, see str.encode()

Returns:¶

converted bytes.

Return type:¶

bytes

octoprint.util.to_native_str(s_or_u: str | bytes, encoding: str = 'utf-8', errors: str = 'strict') → str¶: Deprecated since version 1.8.0: to_native_str is no longer needed, use to_unicode instead

octoprint.util.to_str(s_or_u: str | bytes, encoding: str = 'utf-8', errors: str = 'strict') → bytes¶: Deprecated since version 1.3.11: to_str has been renamed to to_bytes and in a future version will become the new to_unicode

octoprint.util.to_unicode(s_or_u: str | bytes, encoding: str = 'utf-8', errors: str = 'strict') → str¶

Make sure s_or_u is a string (str).

Parameters:¶

s_or_u : str or bytes¶: The value to convert
encoding : str¶: encoding to use if necessary, see bytes.decode()
errors : str¶: error handling to use if necessary, see bytes.decode()

Returns:¶

converted string.

Return type:¶

str

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: str | list[str] | tuple[str], delimiter: bytes = b'\n', buffer_size: int = -1, logged: bool = True, output_timeout: float = 0.5, **kwargs) → tuple[int | None, list[str], list[str]]¶

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: str | list[str] | tuple[str], **kwargs) → tuple[int, list[str], list[str]]¶

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: str | bytes) → str | bytes¶

Removes ANSI control codes from line.

Note: This function also still supports an input of bytes, leading to an output of bytes. This if for reasons of backwards compatibility only, should no longer be used and considered to be deprecated and to be removed in a future version of OctoPrint. A warning will be logged.

Parameters:¶

line : str or bytes¶: the line to process

Returns:¶

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

Changed in version 1.8.0: Usage as clean_ansi(line: bytes) -> bytes is now deprecated and will be removed in a future version of OctoPrint.

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.

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), freebsd (sys.platform = freebsd*) and android (sys.platform = android).

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.