Skip to content

octoprint.util#

This module bundles commonly used utility methods or helper classes that are used in multiple places within OctoPrint's source code.

deprecated = warning_decorator_factory(DeprecationWarning) module-attribute #

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:

Name Type Description Default
message string

The message to include in the deprecation warning.

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

required
includedoc string

Message about the deprecation to include in the wrapped function's docstring.

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

required
since string

Version since when the function was deprecated, must be present for the docstring to get extended.

required

Returns:

Name Type Description
function

The wrapped function with the deprecation warnings in place.

pending_deprecation = warning_decorator_factory(PendingDeprecationWarning) module-attribute #

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:

Name Type Description Default
message string

The message to include in the deprecation warning.

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

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

required
includedoc string

Message about the deprecation to include in the wrapped function's docstring.

required
since string

Version since when the function was deprecated, must be present for the docstring to get extended.

required

Returns:

Name Type Description
function

The wrapped function with the deprecation warnings in place.

variable_deprecated = warning_factory(DeprecationWarning) module-attribute #

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:

Name Type Description Default
message string

The message to include in the deprecation warning.

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

required
since string

Version since when the function was deprecated, must be present for the docstring to get extended.

required

Returns:

Name Type Description
value

The value of the variable with the deprecation warnings in place.

variable_pending_deprecation = warning_factory(PendingDeprecationWarning) module-attribute #

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:

Name Type Description Default
message string

The message to include in the deprecation warning.

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

required
since string

Version since when the function was deprecated, must be present for the docstring to get extended.

required

Returns:

Name Type Description
value

The value of the variable with the deprecation warnings in place.

CaseInsensitiveSet #

Bases: Set

Basic case insensitive set

Any str or unicode values will be stored and compared in lower case. Other value types are left as-is.

RepeatedTimer #

Bases: threading.Thread

This class represents an action that should be run repeatedly in an interval. It is similar to python's own :class: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 :func: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:

.. code-block:: python

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:

.. code-block:: python

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:

Name Type Description Default
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.

required
function callable

The function to call.

required
args list or tuple

The arguments for the function call. Defaults to an empty list.

None
kwargs dict

The keyword arguments for the function call. Defaults to an empty dict.

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

False
condition callable

Condition that needs to be True for loop to continue. Defaults to lambda: True.

None
on_condition_false callable

Callback to call when the timer finishes due to condition becoming false. Will be called before the on_finish callback.

None
on_cancelled callable

Callback to call when the timer finishes due to being cancelled. Will be called before the on_finish callback.

None
on_finish callable

Callback to call when the timer finishes, either due to being cancelled or since the condition became false.

None
daemon bool

daemon flag to set on underlying thread.

True

ResettableTimer #

Bases: threading.Thread

This class represents an action that should be run after a specified amount of time. It is similar to python's own :class: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 :func:cancel method. Resetting the counter can be done with the :func:reset method.

For example:

.. code-block:: python

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:

Name Type Description Default
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.

required
function callable

The function to call.

required
args list or tuple

The arguments for the function call. Defaults to an empty list.

None
kwargs dict

The keyword arguments for the function call. Defaults to an empty dict.

None
on_cancelled callable

Callback to call when the timer finishes due to being cancelled.

None
on_reset callable

Callback to call when the timer is reset.

None
daemon bool

daemon flag to set on underlying thread.

True

chunks(l, n) #

Yield successive n-sized chunks from l.

Taken from http://stackoverflow.com/a/312464/2028598

count(gen) #

Used instead of len(generator), which doesn't work

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:

Name Type Description Default
filename str

The file to deserialize from

required
encoding str

The encoding to use for the file, defaults to utf-8

'utf-8'

Returns:

Type Description

The deserialized data structure

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:

Name Type Description Default
a dict

The dictionary to check for the keys from b.

required
b dict

The dictionary whose keys to check a for.

required

Returns:

Name Type Description
boolean

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

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:

Name Type Description Default
dictionary dict

The dictionary to filter

required
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

required

Returns:

Name Type Description
dict

A shallow copy of the provided dictionary, stripped of the key-value-pairs for which the filter_function returned False

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:

Name Type Description Default
dictionary

the dictionary to flatten

required
prefix

the key prefix, initially an empty string

''
separator

key separator, '.' by default

'.'

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:

Name Type Description Default
a dict

The dictionary to merge b into

required
b dict

The dictionary to merge into a

required
leaf_merger callable

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

None
in_place boolean

If set to True, a will be merged with b in place, meaning a will be modified

False

Returns:

Name Type Description
dict

b deep-merged into a

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:

Name Type Description Default
source dict

Source dictionary

required
target dict

Dictionary to compare to source dictionary and derive diff for

required

Returns:

Name Type Description
dict

The minimal dictionary to deep merge on source to get the same result as deep merging target on source.

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:

Name Type Description Default
a dict

The dictionary to clean against b.

required
b dict

The dictionary containing the key structure to clean from a.

required
Results

filter_non_ascii(line) #

Filter predicate to test if a line contains non ASCII characters.

Parameters:

Name Type Description Default
line string

The line to test

required

Returns:

Name Type Description
boolean

True if the line contains non ASCII characters, False otherwise.

get_bom(filename, encoding) #

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

Parameters:

Name Type Description Default
filename str

The file to check.

required
encoding str

The encoding to check for.

required

Returns:

Type Description

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

get_class(name) #

Retrieves the class object for a given fully qualified class name.

Parameters:

Name Type Description Default
name string

The fully qualified class name, including all modules separated by .

required

Returns:

Name Type Description
type

The class if it could be found.

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:

Name Type Description
string

The formatted exception information.

get_formatted_datetime(d) #

Formats a datetime instance as "YYYY-mm-dd HH:MM" and returns the resulting string.

Parameters:

Name Type Description Default
d datetime.datetime

The datetime instance to format

required

Returns:

Name Type Description
string

The datetime formatted as "YYYY-mm-dd HH:MM"

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:

Name Type Description Default
num int

The byte count to format

required

Returns:

Name Type Description
string

The formatted byte count.

get_formatted_timedelta(d) #

Formats a timedelta instance as "HH:MM:ss" and returns the resulting string.

Parameters:

Name Type Description Default
d datetime.timedelta

The timedelta instance to format

required

Returns:

Name Type Description
string

The timedelta formatted as "HH:MM:ss"

get_fully_qualified_classname(o) #

Returns the fully qualified class name for an object.

Based on https://stackoverflow.com/a/2020083

Parameters:

Name Type Description Default
o

the object of which to determine the fqcn

required

Returns:

Type Description

(str) the fqcn of the object

is_allowed_file(filename, extensions) #

Determines if the provided filename has one of the supplied extensions. The check is done case-insensitive.

Parameters:

Name Type Description Default
filename string

The file name to check against the extensions.

required
extensions list

The extensions to check against, a list of strings

required
Return

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']"

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 :func: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:

Name Type Description Default
filename str

The file to write to

required
data object

The data to serialize

required
encoding str

The encoding to use for the file

'utf-8'
compressed bool

Whether to compress the data before writing it to the file

True

to_bytes(s_or_u, encoding='utf-8', errors='strict') #

Make sure s_or_u is a byte string.

Parameters:

Name Type Description Default
s_or_u str or bytes

The value to convert

required
encoding str

encoding to use if necessary, see :meth:python:str.encode

'utf-8'
errors str

error handling to use if necessary, see :meth:python:str.encode

'strict'

Returns:

Name Type Description
bytes bytes

converted bytes.

to_unicode(s_or_u, encoding='utf-8', errors='strict') #

Make sure s_or_u is a unicode string (str).

Parameters:

Name Type Description Default
s_or_u str or bytes

The value to convert

required
encoding str

encoding to use if necessary, see :meth:python:bytes.decode

'utf-8'
errors str

error handling to use if necessary, see :meth:python:bytes.decode

'strict'

Returns:

Name Type Description
str str

converted string.

Back to top