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 thefunction
will be run again and again, sleeping the statedinterval
in between.RepeatedTimers are started, as with threads, by calling their
start()
method. The timer can be stopped (in between runs) by calling thecancel()
method. The interval the time waited before execution of a loop may not be exactly the same as the interval specified by the user.For example:
def hello(): print("Hello World!") t = RepeatedTimer(1.0, hello) t.start() # prints "Hello World!" every second
Another example with dynamic interval and loop condition:
count = 0 maximum = 5 factor = 1 def interval(): global count global factor return count * factor def condition(): global count global maximum return count <= maximum def hello(): print("Hello World!") global count count += 1 t = RepeatedTimer(interval, hello, run_first=True, condition=condition) t.start() # prints "Hello World!" 5 times, printing the first one # directly, then waiting 1, 2, 3, 4s in between (adaptive interval)
- Parameters
interval (float or callable) – The interval between each
function
call, in seconds. Can also be a callable returning the interval to use, in case the interval is not static.function (callable) – The function to call.
args (list or tuple) – The arguments for the
function
call. Defaults to an empty list.kwargs (dict) – The keyword arguments for the
function
call. Defaults to an empty dict.run_first (boolean) – If set to True, the function will be run for the first time before the first wait period. If set to False (the default), the function will be run for the first time after the first wait period.
condition (callable) – Condition that needs to be True for loop to continue. Defaults to
lambda: True
.on_condition_false (callable) – Callback to call when the timer finishes due to condition becoming false. Will be called before the
on_finish
callback.on_cancelled (callable) – Callback to call when the timer finishes due to being cancelled. Will be called before the
on_finish
callback.on_finish (callable) – Callback to call when the timer finishes, either due to being cancelled or since the condition became false.
daemon (bool) – daemon flag to set on underlying thread.
- run()
Method representing the thread’s activity.
You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.
- class octoprint.util.ResettableTimer(interval, function, args=None, kwargs=None, on_reset=None, on_cancelled=None, daemon=True)
This class represents an action that should be run after a specified amount of time. It is similar to python’s own
threading.Timer
class, with the addition of being able to reset the counter to zero.ResettableTimers are started, as with threads, by calling their
start()
method. The timer can be stopped (in between runs) by calling thecancel()
method. Resetting the counter can be done with thereset()
method.For example:
def hello(): print("Ran hello() at {}").format(time.time()) t = ResettableTimer(60.0, hello) t.start() print("Started at {}").format(time.time()) time.sleep(30) t.reset() print("Reset at {}").format(time.time())
- Parameters
interval (float or callable) – The interval before calling
function
, in seconds. Can also be a callable returning the interval to use, in case the interval is not static.function (callable) – The function to call.
args (list or tuple) – The arguments for the
function
call. Defaults to an empty list.kwargs (dict) – The keyword arguments for the
function
call. Defaults to an empty dict.on_cancelled (callable) – Callback to call when the timer finishes due to being cancelled.
on_reset (callable) – Callback to call when the timer is reset.
daemon (bool) – daemon flag to set on underlying thread.
- run()
Method representing the thread’s activity.
You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.
- exception octoprint.util.TypeAlreadyInQueue(t, *args, **kwargs)
- class octoprint.util.TypedQueue(maxsize=0)
- get(*args, **kwargs)
Remove and return an item from the queue.
If optional args ‘block’ is true and ‘timeout’ is None (the default), block if necessary until an item is available. If ‘timeout’ is a non-negative number, it blocks at most ‘timeout’ seconds and raises the Empty exception if no item was available within that time. Otherwise (‘block’ is false), return an item if one is immediately available, else raise the Empty exception (‘timeout’ is ignored in that case).
- put(item, item_type=None, *args, **kwargs)
Put an item into the queue.
If optional args ‘block’ is true and ‘timeout’ is None (the default), block if necessary until a free slot is available. If ‘timeout’ is a non-negative number, it blocks at most ‘timeout’ seconds and raises the Full exception if no free slot was available within that time. Otherwise (‘block’ is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (‘timeout’ is ignored in that case).
- octoprint.util.chunks(l, n)
Yield successive n-sized chunks from l.
Taken from http://stackoverflow.com/a/312464/2028598
- octoprint.util.count(gen)
Used instead of len(generator), which doesn’t work
- octoprint.util.deprecated(message, stacklevel=1, since=None, includedoc=None, extenddoc=False)
A decorator for deprecated methods. Logs a deprecation warning via Python’s :mod:`warnings module including the supplied
message
. The call stack level used (for adding the source location of the offending call to the warning) can be overridden using the optionalstacklevel
parameter. If bothsince
andincludedoc
are provided, a deprecation warning will also be added to the function’s docstring by providing or extending its__doc__
property.- Parameters
message (string) – The message to include in the deprecation warning.
stacklevel (int) – Stack level for including the caller of the offending method in the logged warning. Defaults to 1, meaning the direct caller of the method. It might make sense to increase this in case of the function call happening dynamically from a fixed position to not shadow the real caller (e.g. in case of overridden
getattr
methods).includedoc (string) – Message about the deprecation to include in the wrapped function’s docstring.
extenddoc (boolean) – If True the original docstring of the wrapped function will be extended by the deprecation message, if False (default) it will be replaced with the deprecation message.
since (string) – Version since when the function was deprecated, must be present for the docstring to get extended.
- Returns
The wrapped function with the deprecation warnings in place.
- Return type
function
- octoprint.util.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.
- octoprint.util.dict_clean(a, b)
Recursively deep-sanitizes
a
based onb
, removing all keys (and associated values) froma
that do not appear inb
.Example:
>>> a = dict(foo="foo", bar="bar", fnord=dict(a=1, b=2, l=["some", "list"])) >>> b = dict(foo=None, fnord=dict(a=None, b=None)) >>> expected = dict(foo="foo", fnord=dict(a=1, b=2)) >>> dict_sanitize(a, b) == expected True >>> dict_clean(a, b) == expected True
- Parameters
- Results:
dict: A new dict based on
a
with all keys (and corresponding values) found inb
removed.
- octoprint.util.dict_contains_keys(keys, dictionary)
Recursively deep-checks if
dictionary
contains all keys found inkeys
.Example:
>>> positive = dict(foo="some_other_bar", fnord=dict(b=100)) >>> negative = dict(foo="some_other_bar", fnord=dict(b=100, d=20)) >>> dictionary = dict(foo="bar", fnord=dict(a=1, b=2, c=3)) >>> dict_contains_keys(positive, dictionary) True >>> dict_contains_keys(negative, dictionary) False
- octoprint.util.dict_filter(dictionary, filter_function)
Filters a dictionary with the provided filter_function
Example:
>>> data = dict(key1="value1", key2="value2", other_key="other_value", foo="bar", bar="foo") >>> dict_filter(data, lambda k, v: k.startswith("key")) == dict(key1="value1", key2="value2") True >>> dict_filter(data, lambda k, v: v.startswith("value")) == dict(key1="value1", key2="value2") True >>> dict_filter(data, lambda k, v: k == "foo" or v == "foo") == dict(foo="bar", bar="foo") True >>> dict_filter(data, lambda k, v: False) == dict() True >>> dict_filter(data, lambda k, v: True) == data True >>> dict_filter(None, lambda k, v: True) Traceback (most recent call last): ... AssertionError >>> dict_filter(data, None) Traceback (most recent call last): ... AssertionError
- Parameters
dictionary (dict) – The dictionary to filter
filter_function (callable) – The filter function to apply, called with key and value of an entry in the dictionary, must return
True
for values to keep andFalse
for values to strip
- Returns
- A shallow copy of the provided dictionary, stripped of the key-value-pairs
for which the
filter_function
returnedFalse
- Return type
- octoprint.util.dict_flatten(dictionary, prefix='', separator='.')
Flatten a dictionary.
- Example::
>>> data = {'a': {'a1': 'a1', 'a2': 'a2'}, 'b': 'b'} >>> expected = {'a.a1': 'a1', 'a.a2': 'a2', 'b': 'b'} >>> actual = dict_flatten(data) >>> shared = {(k, actual[k]) for k in actual if k in expected and actual[k] == expected[k]} >>> len(shared) == len(expected) True
- Parameters
dictionary – the dictionary to flatten
prefix – the key prefix, initially an empty string
separator – key separator, ‘.’ by default
Returns: a flattened dict
- octoprint.util.dict_merge(a, b, leaf_merger=None, in_place=False)
Recursively deep-merges two dictionaries.
Based on https://www.xormedia.com/recursively-merge-dictionaries-in-python/
Example:
>>> a = dict(foo="foo", bar="bar", fnord=dict(a=1)) >>> b = dict(foo="other foo", fnord=dict(b=2, l=["some", "list"])) >>> expected = dict(foo="other foo", bar="bar", fnord=dict(a=1, b=2, l=["some", "list"])) >>> dict_merge(a, b) == expected True >>> dict_merge(a, None) == a True >>> dict_merge(None, b) == b True >>> dict_merge(None, None) == dict() True >>> def leaf_merger(a, b): ... if isinstance(a, list) and isinstance(b, list): ... return a + b ... raise ValueError() >>> result = dict_merge(dict(l1=[3, 4], l2=[1], a="a"), dict(l1=[1, 2], l2="foo", b="b"), leaf_merger=leaf_merger) >>> result.get("l1") == [3, 4, 1, 2] True >>> result.get("l2") == "foo" True >>> result.get("a") == "a" True >>> result.get("b") == "b" True >>> c = dict(foo="foo") >>> dict_merge(c, {"bar": "bar"}) is c False >>> dict_merge(c, {"bar": "bar"}, in_place=True) is c True
- Parameters
- Returns
b
deep-merged intoa
- Return type
- octoprint.util.dict_minimal_mergediff(source, target)
Recursively calculates the minimal dict that would be needed to be deep merged with a in order to produce the same result as deep merging a and b.
Example:
>>> a = dict(foo=dict(a=1, b=2), bar=dict(c=3, d=4)) >>> b = dict(bar=dict(c=3, d=5), fnord=None) >>> c = dict_minimal_mergediff(a, b) >>> c == dict(bar=dict(d=5), fnord=None) True >>> dict_merge(a, c) == dict_merge(a, b) True
- octoprint.util.dict_sanitize(a, b)
Recursively deep-sanitizes
a
based onb
, removing all keys (and associated values) froma
that do not appear inb
.Example:
>>> a = dict(foo="foo", bar="bar", fnord=dict(a=1, b=2, l=["some", "list"])) >>> b = dict(foo=None, fnord=dict(a=None, b=None)) >>> expected = dict(foo="foo", fnord=dict(a=1, b=2)) >>> dict_sanitize(a, b) == expected True >>> dict_clean(a, b) == expected True
- Parameters
- Results:
dict: A new dict based on
a
with all keys (and corresponding values) found inb
removed.
- class octoprint.util.fallback_dict(custom, *fallbacks)
- items() a set-like object providing a view on D's items
- keys() a set-like object providing a view on D's keys
- values() an object providing a view on D's values
- octoprint.util.filter_non_ascii(line)
Filter predicate to test if a line contains non ASCII characters.
- Parameters
line (string) – The line to test
- Returns
True if the line contains non ASCII characters, False otherwise.
- Return type
boolean
- octoprint.util.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
- Raises
- 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 suppliedextensions
. The check is done case-insensitive.- Parameters
filename (string) – The file name to check against the extensions.
extensions (list) – The extensions to check against, a list of strings
- Returns
True if the file name’s extension matches one of the allowed extensions, False otherwise.
- Return type
boolean
- octoprint.util.pending_deprecation(message, stacklevel=1, since=None, includedoc=None, extenddoc=False)
A decorator for methods pending deprecation. Logs a pending deprecation warning via Python’s :mod:`warnings module including the supplied
message
. The call stack level used (for adding the source location of the offending call to the warning) can be overridden using the optionalstacklevel
parameter. If bothsince
andincludedoc
are provided, a deprecation warning will also be added to the function’s docstring by providing or extending its__doc__
property.- Parameters
message (string) – The message to include in the deprecation warning.
stacklevel (int) – Stack level for including the caller of the offending method in the logged warning. Defaults to 1, meaning the direct caller of the method. It might make sense to increase this in case of the function call happening dynamically from a fixed position to not shadow the real caller (e.g. in case of overridden
getattr
methods).extenddoc (boolean) – If True the original docstring of the wrapped function will be extended by the deprecation message, if False (default) it will be replaced with the deprecation message.
includedoc (string) – Message about the deprecation to include in the wrapped function’s docstring.
since (string) – Version since when the function was deprecated, must be present for the docstring to get extended.
- Returns
The wrapped function with the deprecation warnings in place.
- Return type
function
- octoprint.util.pp(value)
>>> pp(dict()) 'dict()' >>> pp(dict(a=1, b=2, c=3)) 'dict(a=1, b=2, c=3)' >>> pp(set()) 'set()' >>> pp({"a", "b"}) "{'a', 'b'}" >>> pp(["a", "b", "d", "c"]) "['a', 'b', 'd', 'c']" >>> pp(("a", "b", "d", "c")) "('a', 'b', 'd', 'c')" >>> pp("foo") "'foo'" >>> pp([dict(a=1, b=2), {"a", "c", "b"}, (1, 2), None, 1, True, "foo"]) "[dict(a=1, b=2), {'a', 'b', 'c'}, (1, 2), None, 1, True, 'foo']"
- octoprint.util.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.
- octoprint.util.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
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
- octoprint.util.to_native_str(s_or_u: Union[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: Union[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: Union[str, bytes], encoding: str = 'utf-8', errors: str = 'strict') str
Make sure
s_or_u
is a string (str).- Parameters
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
- octoprint.util.variable_deprecated(message, stacklevel=1, since=None, includedoc=None, extenddoc=False)
A function for deprecated variables. Logs a deprecation warning via Python’s :mod:`warnings module including the supplied
message
. The call stack level used (for adding the source location of the offending call to the warning) can be overridden using the optionalstacklevel
parameter.- Parameters
message (string) – The message to include in the deprecation warning.
stacklevel (int) – Stack level for including the caller of the offending method in the logged warning. Defaults to 1, meaning the direct caller of the method. It might make sense to increase this in case of the function call happening dynamically from a fixed position to not shadow the real caller (e.g. in case of overridden
getattr
methods).since (string) – Version since when the function was deprecated, must be present for the docstring to get extended.
- Returns
The value of the variable with the deprecation warnings in place.
- Return type
value
- octoprint.util.variable_pending_deprecation(message, stacklevel=1, since=None, includedoc=None, extenddoc=False)
A decorator for variables pending deprecation. Logs a pending deprecation warning via Python’s :mod:`warnings module including the supplied
message
. The call stack level used (for adding the source location of the offending call to the warning) can be overridden using the optionalstacklevel
parameter.- Parameters
message (string) – The message to include in the deprecation warning.
stacklevel (int) – Stack level for including the caller of the offending method in the logged warning. Defaults to 1, meaning the direct caller of the method. It might make sense to increase this in case of the function call happening dynamically from a fixed position to not shadow the real caller (e.g. in case of overridden
getattr
methods).since (string) – Version since when the function was deprecated, must be present for the docstring to get extended.
- Returns
The value of the variable with the deprecation warnings in place.
- Return type
value
octoprint.util.commandline
- class octoprint.util.commandline.CommandlineCaller
The CommandlineCaller is a utility class that allows running command line commands while logging their stdout and stderr via configurable callback functions.
Callbacks are expected to have a signature matching
def callback(*lines): do_something_with_the_passed_lines()
The class utilizes sarge underneath.
Example:
from octoprint.util.commandline import CommandLineCaller, CommandLineError def log(prefix, *lines): for line in lines: print("{} {}".format(prefix, line)) def log_stdout(*lines): log(">>>", *lines) def log_stderr(*lines): log("!!!", *lines) def log_call(*lines) log("---", *lines) caller = CommandLineCaller() caller.on_log_call = log_call caller.on_log_stdout = log_stdout caller.on_log_stderr = log_stderr try: caller.checked_call(["some", "command", "with", "parameters"]) except CommandLineError as err: print("Command returned {}".format(err.returncode)) else: print("Command finished successfully")
- call(command: Union[str, List[str], Tuple[str]], delimiter: bytes = b'\n', buffer_size: int = -1, logged: bool = True, output_timeout: float = 0.5, **kwargs) Tuple[Optional[int], List[str], List[str]]
Calls a command
- checked_call(command: Union[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
- Returns
(tuple) a 3-tuple of return code, full stdout and full stderr output
- Raises
- on_log_call
Callback for the called command line
- on_log_stderr
Callback for stderr output
- on_log_stdout
Callback for stdout output
- exception octoprint.util.commandline.CommandlineError(returncode, stdout, stderr)
Raised by
checked_call()
on non zero return codes
- class octoprint.util.commandline.DelimiterCapture(delimiter=b'\n', *args, **kwargs)
- octoprint.util.commandline.clean_ansi(line: Union[str, bytes]) Union[str, bytes]
Removes ANSI control codes from
line
.Note: This function also still supports an input of
bytes
, leading to anoutput
ofbytes
. 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
- 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
) andfreebsd
(sys.platform
=freebsd*
).- Returns
(str) mapped OS identifier
- octoprint.util.platform.is_os_compatible(compatibility_entries, current_os=None)
Tests if the
current_os
orsys.platform
are blacklisted or whitelisted incompatibility_entries
- Returns
(bool) True if the os is compatible, False otherwise
- octoprint.util.platform.set_close_exec(handle)
Set
close_exec
flag on handle, if supported by the OS.