octoprint.util

class octoprint.util.ConnectivityChecker(interval, host, port, enabled=True, on_change=None)

Regularly checks for online connectivity.

Tries to open a connection to the provided host and port every interval seconds and sets the online status accordingly.

check_immediately()

Check immediately and return result.

enabled

Whether the check is enabled or not.

host

DNS host to query.

interval

Interval between consecutive automatic checks.

online

Current online status, True if online, False if offline.

port

DNS port to query.

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.
octoprint.util.address_for_client(host, port, timeout=3.05)

Determines the address of the network interface on this host needed to connect to the indicated client host and port.

octoprint.util.chunks(l, n)

Yield successive n-sized chunks from l.

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

octoprint.util.deprecated(message, stacklevel=1, since=None, includedoc=None, extenddoc=False)

A decorator for deprecated methods. Logs a deprecation warning via Python’s :mod:`warnings module including the supplied message. The call stack level used (for adding the source location of the offending call to the warning) can be overridden using the optional stacklevel parameter. If both since and includedoc are provided, a deprecation warning will also be added to the function’s docstring by providing or extending its __doc__ property.

Parameters:
  • message (string) – The message to include in the deprecation warning.
  • stacklevel (int) – Stack level for including the caller of the offending method in the logged warning. Defaults to 1, meaning the direct caller of the method. It might make sense to increase this in case of the function call happening dynamically from a fixed position to not shadow the real caller (e.g. in case of overridden getattr methods).
  • includedoc (string) – Message about the deprecation to include in the wrapped function’s docstring.
  • extenddoc (boolean) – If True the original docstring of the wrapped function will be extended by the deprecation message, if False (default) it will be replaced with the deprecation message.
  • since (string) – Version since when the function was deprecated, must be present for the docstring to get extended.
Returns:

The wrapped function with the deprecation warnings in place.

Return type:

function

octoprint.util.dict_clean(*args, **kwargs)

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

Example:

>>> a = dict(foo="foo", bar="bar", fnord=dict(a=1, b=2, l=["some", "list"]))
>>> b = dict(foo=None, fnord=dict(a=None, b=None))
>>> expected = dict(foo="foo", fnord=dict(a=1, b=2))
>>> dict_sanitize(a, b) == expected
True
>>> dict_clean(a, b) == expected
True
Parameters:
  • a (dict) – The dictionary to clean against b.
  • b (dict) – The dictionary containing the key structure to clean from a.
Results:
dict: A new dict based on a with all keys (and corresponding values) found in b removed.
octoprint.util.dict_contains_keys(keys, dictionary)

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

Example:

>>> positive = dict(foo="some_other_bar", fnord=dict(b=100))
>>> negative = dict(foo="some_other_bar", fnord=dict(b=100, d=20))
>>> dictionary = dict(foo="bar", fnord=dict(a=1, b=2, c=3))
>>> dict_contains_keys(positive, dictionary)
True
>>> dict_contains_keys(negative, dictionary)
False
Parameters:
  • a (dict) – The dictionary to check for the keys from b.
  • b (dict) – The dictionary whose keys to check a for.
Returns:

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

Return type:

boolean

octoprint.util.dict_filter(dictionary, filter_function)

Filters a dictionary with the provided filter_function

Example:

>>> data = dict(key1="value1", key2="value2", other_key="other_value", foo="bar", bar="foo")
>>> dict_filter(data, lambda k, v: k.startswith("key")) == dict(key1="value1", key2="value2")
True
>>> dict_filter(data, lambda k, v: v.startswith("value")) == dict(key1="value1", key2="value2")
True
>>> dict_filter(data, lambda k, v: k == "foo" or v == "foo") == dict(foo="bar", bar="foo")
True
>>> dict_filter(data, lambda k, v: False) == dict()
True
>>> dict_filter(data, lambda k, v: True) == data
True
>>> dict_filter(None, lambda k, v: True)
Traceback (most recent call last):
    ...
AssertionError
>>> dict_filter(data, None)
Traceback (most recent call last):
    ...
AssertionError
Parameters:
  • dictionary (dict) – The dictionary to filter
  • filter_function (callable) – The filter function to apply, called with key and value of an entry in the dictionary, must return True for values to keep and False for values to strip
Returns:

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

for which the filter_function returned False

Return type:

dict

octoprint.util.dict_merge(a, b)

Recursively deep-merges two dictionaries.

Taken from 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
Parameters:
  • a (dict) – The dictionary to merge b into
  • b (dict) – The dictionary to merge into a
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.
octoprint.util.filter_non_ascii(line)

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

Parameters:line (string) – The line to test
Returns:True if the line contains non ASCII characters, False otherwise.
Return type:boolean
octoprint.util.find_collision_free_name(filename, extension, existing_filenames, max_power=2)

Tries to find a collision free translation of “<filename>.<extension>” to the 8.3 DOS compatible format, preventing collisions with any of the existing_filenames.

First strips all of ."/\[]:;=, from the filename and extensions, converts them to lower case and truncates the extension to a maximum length of 3 characters.

If the filename is already equal or less than 8 characters in length after that procedure and “<filename>.<extension>” are not contained in the existing_files, that concatenation will be returned as the result.

If not, the following algorithm will be applied to try to find a collision free name:

set counter := power := 1
while counter < 10^max_power:
    set truncated := substr(filename, 0, 6 - power + 1) + "~" + counter
    set result := "<truncated>.<extension>"
    if result is collision free:
        return result
    counter++
    if counter >= 10 ** power:
        power++
raise ValueError

This will basically – for a given original filename of some_filename and an extension of gco – iterate through names of the format some_f~1.gco, some_f~2.gco, …, some_~10.gco, some_~11.gco, …, <prefix>~<n>.gco for n less than 10 ^ max_power, returning as soon as one is found that is not colliding.

Parameters:
  • filename (string) – The filename without the extension to convert to 8.3.
  • extension (string) – The extension to convert to 8.3 – will be truncated to 3 characters if it’s longer than that.
  • existing_filenames (list) – A list of existing filenames to prevent name collisions with.
  • max_power (int) – Limits the possible attempts of generating a collision free name to 10 ^ max_power variations. Defaults to 2, so the name generation will maximally reach <name>~99.<ext> before aborting and raising an exception.
Returns:

A 8.3 representation of the provided original filename, ensured to not collide with the provided

existing_filenames

Return type:

string

Raises:

ValueError – No collision free name could be found.

Examples

>>> find_collision_free_name("test1234", "gco", [])
u'test1234.gco'
>>> find_collision_free_name("test1234", "gcode", [])
u'test1234.gco'
>>> find_collision_free_name("test12345", "gco", [])
u'test12~1.gco'
>>> find_collision_free_name("test 123", "gco", [])
u'test_123.gco'
>>> find_collision_free_name("test1234", "g o", [])
u'test1234.g_o'
>>> find_collision_free_name("test12345", "gco", ["test12~1.gco"])
u'test12~2.gco'
>>> many_files = ["test12~{}.gco".format(x) for x in range(10)[1:]]
>>> find_collision_free_name("test12345", "gco", many_files)
u'test1~10.gco'
>>> many_more_files = many_files + ["test1~{}.gco".format(x) for x in range(10, 99)]
>>> find_collision_free_name("test12345", "gco", many_more_files)
u'test1~99.gco'
>>> many_more_files_plus_one = many_more_files + ["test1~99.gco"]
>>> find_collision_free_name("test12345", "gco", many_more_files_plus_one)
Traceback (most recent call last):
...
ValueError: Can't create a collision free filename
>>> find_collision_free_name("test12345", "gco", many_more_files_plus_one, max_power=3)
u'test~100.gco'
octoprint.util.get_class(name)

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

Parameters:name (string) – The fully qualified class name, including all modules separated by .
Returns:The class if it could be found.
Return type:type
Raises:ImportError
octoprint.util.get_dos_filename(input, existing_filenames=None, extension=None, whitelisted_extensions=None, **kwargs)

Converts the provided input filename to a 8.3 DOS compatible filename. If existing_filenames is provided, the conversion result will be guaranteed not to collide with any of the filenames provided thus.

Uses find_collision_free_name() internally.

Parameters:
  • input (string) – The original filename incl. extension to convert to the 8.3 format.
  • existing_filenames (list) – A list of existing filenames with which the generated 8.3 name must not collide. Optional.
  • extension (string) – The .3 file extension to use for the generated filename. If not provided, the extension of the provided filename will simply be truncated to 3 characters.
  • whitelisted_extensions (list) – A list of extensions on input that will be left as-is instead of exchanging for extension.
  • kwargs (dict) – Additional keyword arguments to provide to find_collision_free_name().
Returns:

A 8.3 compatible translation of the original filename, not colliding with the optionally provided

existing_filenames and with the provided extension or the original extension shortened to a maximum of 3 characters.

Return type:

string

Raises:

ValueError – No 8.3 compatible name could be found that doesn’t collide with the provided existing_filenames.

Examples

>>> get_dos_filename("test1234.gco")
u'test1234.gco'
>>> get_dos_filename("test1234.gcode")
u'test1234.gco'
>>> get_dos_filename("test12345.gco")
u'test12~1.gco'
>>> get_dos_filename("test1234.fnord", extension="gco")
u'test1234.gco'
>>> get_dos_filename("auto0.g", extension="gco")
u'auto0.gco'
>>> get_dos_filename("auto0.g", extension="gco", whitelisted_extensions=["g"])
u'auto0.g'
>>> get_dos_filename(None)
>>> get_dos_filename("foo")
u'foo'
octoprint.util.get_exception_string()

Retrieves the exception info of the last raised exception and returns it as a string formatted as <exception type>: <exception message> @ <source file>:<function name>:<line number>.

Returns:The formatted exception information.
Return type:string
octoprint.util.get_formatted_datetime(d)

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

Parameters:d (datetime.datetime) – The datetime instance to format
Returns:The datetime formatted as “YYYY-mm-dd HH:MM”
Return type:string
octoprint.util.get_formatted_size(num)

Formats the given byte count as a human readable rounded size expressed in the most pressing unit among B(ytes), K(ilo)B(ytes), M(ega)B(ytes), G(iga)B(ytes) and T(era)B(ytes), with one decimal place.

Based on http://stackoverflow.com/a/1094933/2028598

Parameters:num (int) – The byte count to format
Returns:The formatted byte count.
Return type:string
octoprint.util.get_formatted_timedelta(d)

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

Parameters:d (datetime.timedelta) – The timedelta instance to format
Returns:The timedelta formatted as “HH:MM:ss”
Return type:string
octoprint.util.get_free_bytes(*args, **kwargs)

Deprecated since version 1.2.5: Replaced by psutil.disk_usage.

octoprint.util.interface_addresses(family=None)

Retrieves all of the host’s network interface addresses.

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.server_reachable(host, port, timeout=3.05, proto='tcp', source=None)

Checks if a server is reachable

Parameters:
  • host (str) – host to check against
  • port (int) – port to check against
  • timeout (float) – timeout for check
  • proto (str) – tcp or udp
  • source (str) – optional, socket used for check will be bound against this address if provided
Returns:

True if a connection to the server could be opened, False otherwise

Return type:

boolean

octoprint.util.silent_remove(file)

Silently removes a file. Does not raise an error if the file doesn’t exist.

Parameters:file (string) – The path of the file to be removed
octoprint.util.to_str(s_or_u, encoding='utf-8', errors='strict')

Make sure s_or_u is a str.

octoprint.util.to_unicode(s_or_u, encoding='utf-8', errors='strict')

Make sure s_or_u is a unicode string.