octoprint.util#

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

Changed in version 1.8.0

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

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:

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:

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:

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:

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:

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:

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:

value –

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

`CaseInsensitiveSet(*args)` #

Bases: Set

Basic case insensitive set

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

`RepeatedTimer(interval, function, args = None, kwargs = None, run_first = False, condition = None, on_condition_false = None, on_cancelled = None, on_finish = None, daemon = True)` #

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:

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.

`ResettableTimer(interval, function, args = None, kwargs = None, on_reset = None, on_cancelled = None, daemon = True)` #

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:

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.

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

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

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

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:

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:

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:

dictionary –

the dictionary to flatten
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:

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:

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:

source (dict) –

Source dictionary
target (dict) –

Dictionary to compare to source dictionary and derive diff for

Returns:

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:

a (dict) –

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

The dictionary containing the key structure to clean from a.

Results

`filter_non_ascii(line)` #

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

Parameters:

line (string) –

The line to test

Returns:

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:

filename (str) –

The file to check.
encoding (str) –

The encoding to check for.

Returns:

–

(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 (string) –

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

Returns:

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:

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:

d (datetime.datetime) –

The datetime instance to format

Returns:

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:

num (int) –

The byte count to format

Returns:

string –

The formatted byte count.

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

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:

o –

the object of which to determine the fqcn

Returns:

–

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

filename (string) –

The file name to check against the extensions.
extensions (list) –

The extensions to check against, a list of strings

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:

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

`to_bytes(s_or_u: Union[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 :meth:python:str.encode
errors (str) –

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

Returns:

bytes( bytes ) –

converted bytes.

`to_unicode(s_or_u: Union[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 :meth:python:bytes.decode
errors (str) –

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

Returns:

str( str ) –

converted string.

octoprint.util#

deprecated = warning_decorator_factory(DeprecationWarning) module-attribute #

pending_deprecation = warning_decorator_factory(PendingDeprecationWarning) module-attribute #

variable_deprecated = warning_factory(DeprecationWarning) module-attribute #

variable_pending_deprecation = warning_factory(PendingDeprecationWarning) module-attribute #

CaseInsensitiveSet(*args) #

RepeatedTimer(interval, function, args = None, kwargs = None, run_first = False, condition = None, on_condition_false = None, on_cancelled = None, on_finish = None, daemon = True) #

ResettableTimer(interval, function, args = None, kwargs = None, on_reset = None, on_cancelled = None, daemon = True) #

chunks(l, n) #

count(gen) #

deserialize(filename, encoding = 'utf-8') #

dict_contains_keys(keys, dictionary) #

dict_filter(dictionary, filter_function) #

dict_flatten(dictionary, prefix = '', separator = '.') #

dict_merge(a, b, leaf_merger = None, in_place = False) #

dict_minimal_mergediff(source, target) #

dict_sanitize(a, b) #

filter_non_ascii(line) #

get_bom(filename, encoding) #

get_class(name) #

get_exception_string(fmt = "{type}: '{message}' @ {file}:{function}:{line}") #

get_formatted_datetime(d) #

get_formatted_size(num) #

get_formatted_timedelta(d) #

get_fully_qualified_classname(o) #

is_allowed_file(filename, extensions) #

pp(value) #

serialize(filename, data, encoding = 'utf-8', compressed = True) #

to_bytes(s_or_u: Union[str, bytes], encoding: str = 'utf-8', errors: str = 'strict') -> bytes #

to_unicode(s_or_u: Union[str, bytes], encoding: str = 'utf-8', errors: str = 'strict') -> str #

`deprecated = warning_decorator_factory(DeprecationWarning)` `module-attribute` #

`pending_deprecation = warning_decorator_factory(PendingDeprecationWarning)` `module-attribute` #

`variable_deprecated = warning_factory(DeprecationWarning)` `module-attribute` #

`variable_pending_deprecation = warning_factory(PendingDeprecationWarning)` `module-attribute` #

`CaseInsensitiveSet(*args)` #

`RepeatedTimer(interval, function, args = None, kwargs = None, run_first = False, condition = None, on_condition_false = None, on_cancelled = None, on_finish = None, daemon = True)` #

`ResettableTimer(interval, function, args = None, kwargs = None, on_reset = None, on_cancelled = None, daemon = True)` #

`chunks(l, n)` #

`count(gen)` #

`deserialize(filename, encoding = 'utf-8')` #

`dict_contains_keys(keys, dictionary)` #

`dict_filter(dictionary, filter_function)` #

`dict_flatten(dictionary, prefix = '', separator = '.')` #

`dict_merge(a, b, leaf_merger = None, in_place = False)` #

`dict_minimal_mergediff(source, target)` #

`dict_sanitize(a, b)` #

`filter_non_ascii(line)` #

`get_bom(filename, encoding)` #

`get_class(name)` #

`get_exception_string(fmt = "{type}: '{message}' @ {file}:{function}:{line}")` #

`get_formatted_datetime(d)` #

`get_formatted_size(num)` #

`get_formatted_timedelta(d)` #

`get_fully_qualified_classname(o)` #

`is_allowed_file(filename, extensions)` #

`pp(value)` #

`serialize(filename, data, encoding = 'utf-8', compressed = True)` #

`to_bytes(s_or_u: Union[str, bytes], encoding: str = 'utf-8', errors: str = 'strict') -> bytes` #

`to_unicode(s_or_u: Union[str, bytes], encoding: str = 'utf-8', errors: str = 'strict') -> str` #