Difference Functions

pydoover.utils.apply_diff(data: dict[str, Any], diff: dict[str, Any], do_delete: bool = True, clone: bool = True)[source]

Apply a doover compatible diff to a JSON / dict object.

Returns a new object with the diff applied.

To modify the object in-place, pass clone=False.

pydoover.utils.generate_diff(old, new, do_delete: bool = True)[source]

Generate a doover compatible diff between two JSON / dict objects.

A diff will contain all keys that are different between the two objects.

Any keys that are in the old object but not in the new object will be set to None if do_delete is True.

Kalman Filters

pydoover.utils.apply_kalman_filter(initial_estimate=None, initial_error_estimate=None, process_variance=None, outlier_protection=None, outlier_threshold=None, outlier_variance_multiplier=None)[source]

A decorator to apply a Kalman filter to the return value of a function The function should return a single value (e.g., a sensor reading) See below for an example of how to use this decorator

Parameters:

  • initial_estimate

  • initial_error_estimate

  • process_variance

  • outlier_protection

  • outlier_threshold

  • outlier_variance_multiplier

pydoover.utils.apply_async_kalman_filter(initial_estimate=None, initial_error_estimate=None, process_variance=None, outlier_protection=None, outlier_threshold=None, outlier_variance_multiplier=None)[source]

PID Controllers

class pydoover.utils.PID(Kp, Ki, Kd, setpoint=0, output_limits=(None, None))[source]
reset()[source]

Reset the internal state of the PID controller.

set_integral_output(integral_output)[source]

Initialise the integral output for a desired output value.

Parameters:

integral – The integral integral_output value

set_last_error(error)[source]

Set the last error value.

Parameters:

error – The last error value

set_last_output(output)[source]

Set the last output value.

Parameters:

output – The last output value

set_output_limits(min_output, max_output)[source]

Set the minimum and maximum output limits.

Parameters:

  • min_output – Minimum limit

  • max_output – Maximum limit

set_setpoint(setpoint)[source]

Set a new target value for the PID to reach.

Parameters:

setpoint – The target value

update(feedback_value, dt=None)[source]

Update the PID loop with the current feedback value.

Parameters:

  • feedback_value – The current value from the process

  • dt – Optional time interval. If not provided, it’s calculated internally.

Returns:

The control output

Miscellaneous

pydoover.utils.map_reading(in_val, output_values, raw_readings=[4, 20], ignore_below=3)[source]

Map a reading to a value in a range

pydoover.utils.find_object_with_key(obj: dict[Any, Any], key_to_find: str) → Any | None[source]

Iteratively searches through a dictionary (JSON object) and returns the value for the specified key.

Parameters:

  • obj (dict) – The JSON object (dictionary) to search through.

  • key_to_find (str) – The key to search for.

Returns:

The object containing the key, or None if the key is not found.

Return type:

Any

pydoover.utils.find_path_to_key(obj: dict[Any, Any], key_to_find: str) → str | None[source]

Iteratively searches through a dictionary (JSON object) and returns the path to the specified key.

Parameters:

  • obj (dict) – The JSON object (dictionary) to search through.

  • key_to_find (str) – The key to search for.

Returns:

The path to the key, or None if the key is not found.

Return type:

str, optional

pydoover.utils.maybe_async()[source]

Wrapper to allow both a sync and async variation on the same function to provide a unified interface to the user.

This is useful when writing library functions for both a sync and async context.

It assumes you have a variation of your function with the same signature suffixed with _async.

Examples

A simple example:

class MyClass:
    @maybe_async()
    def my_function(self, value: str):
        print("This is the sync version of my_function")
        return "sync result"

    async def my_function_async(self, value: str):
        print("This is the async version of my_function")
        return "async result"

If the user was running an asynchronous main loop, my_function would be invoked as follows:

class MyApp(Application):
    async def main_loop(self):
        result = await obj.my_function("test")
        print(result)  # This would print "async result"

However, if the main loop was synchronous, it would invoke the sync version of the function:

class MyApp(Application):
    def main_loop(self):
        result = self.my_function("test")
        print(result)  # This would print "sync result"
pydoover.utils.wrap_try_except(func, *args, **kwargs)[source]

Wrapper function to catch exceptions and log them. This does not propagate the exception.

async pydoover.utils.wrap_try_except_async(func, *args, **kwargs)[source]

Wrapper function to catch exceptions in an async function and log them. This does not propagate the exception.

async pydoover.utils.call_maybe_async(func, *args, as_task: bool = False, in_executor: bool = True, **kwargs)[source]

Helper function to call a function that may be either synchronous or asynchronous.

Parameters:

  • func (callable) – The function to call, which may be synchronous or asynchronous.

  • *args – Arguments to pass to the function.

  • as_task (bool) – If True, the function will be called as an asyncio task. Default is False.

  • in_executor (bool) – If True, the function will be run in an executor if it is not a coroutine function. Default is True.

  • **kwargs – Any kwargs to pass to the function.

pydoover.utils.on_change(callback, name=None)[source]

A decorator that triggers a callback when the output of the decorated function changes.

The callback is called with four arguments:

  • new_result: The new output of the function.

  • old_result: The previous output (or None if this is the first call).

  • is_first: A boolean indicating if this is the first time the function has returned a value.

  • change_detector_name: The optional name identifying this change detector.

Parameters:

  • callback – The callback function to trigger, or a string indicating the name of an instance attribute.

  • name – An optional name for this change_detector_name.

Examples

A simple usage example:

class MyClass:
    def __init__(self):
        self.last = None

    def my_callback(self, new_result, old_result, is_first, change_detector_name):
        if is_first:
            print(f"{change_detector_name} has returned a value for the first time: {new_result}")
        else:
            print(f"{change_detector_name} has changed from {old_result} to {new_result}")

    @on_change("my_callback", name="my_function")
    def my_function(self):
        import random
        return random.randint(0, 100)