Skip to content

Async Helpers

pydoover provides a set of utility functions for working with mixed sync/async codebases, handling exceptions gracefully, detecting value changes, and mapping sensor readings. These are available from pydoover.utils.

Async Context Detection

get_is_async

Detects whether code is currently running inside an async event loop:

from pydoover.utils import get_is_async

if get_is_async():
    print("Running in an async context")
else:
    print("Running synchronously")

You can also pass an explicit boolean to override the auto-detection:

# Force a specific result regardless of the actual context
is_async = get_is_async(is_async=True)

This is used internally by the maybe_async decorator to determine which implementation to call.

Sync/Async Dual Implementation

maybe_async

The @maybe_async() decorator lets you write a single method name that automatically dispatches to either a sync or async implementation depending on the runtime context. Define the sync version normally and create an async counterpart with the _async suffix:

from pydoover.utils import maybe_async

class SensorReader:
    def __init__(self, is_async=False):
        self._is_async = is_async

    @maybe_async()
    def read_value(self):
        """Synchronous implementation."""
        return self._fetch_sync()

    async def read_value_async(self):
        """Async implementation, called automatically in async contexts."""
        return await self._fetch_async()

When self._is_async is True, calling reader.read_value() returns a coroutine from read_value_async() instead. When False, it calls the sync version directly.

To force the sync version even in an async context, pass run_sync=True:

# Always use the sync implementation
result = reader.read_value(run_sync=True)
Information Circle

The maybe_async decorator expects self as the first argument and checks the _is_async attribute on the instance. It is designed for use on class methods, not standalone functions.

Universal Function Caller

call_maybe_async

call_maybe_async calls any function -- sync or async -- from an async context. It handles coroutines, executor offloading, and task creation:

from pydoover.utils import call_maybe_async

# Call a sync function from async code (runs in executor by default)
result = await call_maybe_async(my_sync_function, arg1, arg2)

# Call an async function normally
result = await call_maybe_async(my_async_function, arg1, arg2)

# Run as a background task (returns immediately)
task = await call_maybe_async(my_async_function, arg1, as_task=True)

# Call sync function without executor (blocks the event loop)
result = await call_maybe_async(my_sync_function, arg1, in_executor=False)
ParameterTypeDefaultDescription
funccallable--The function to call.
*argsPositional arguments passed to the function.
as_taskboolFalseIf True, returns an asyncio.Task instead of awaiting the result.
in_executorboolTrueIf True and the function is synchronous, runs it in the default executor to avoid blocking the event loop.
**kwargsKeyword arguments passed to the function.
Warning

Keyword arguments are not supported when running via the executor (in_executor=True with a sync function). A warning is logged if kwargs are provided in this case.

Exception Wrappers

wrap_try_except

Calls a function and catches any exception, logging it instead of propagating. Returns the function's result on success, or None on failure:

from pydoover.utils import wrap_try_except

# If process_data raises, the exception is logged and None is returned
result = wrap_try_except(process_data, raw_input)

wrap_try_except_async

The async equivalent:

from pydoover.utils import wrap_try_except_async

result = await wrap_try_except_async(async_process_data, raw_input)

These wrappers are useful for fire-and-forget operations where you want to log errors but not crash the application.

Change Detection

on_change

The @on_change(callback) decorator triggers a callback whenever the decorated function's return value differs from its previous return value. This is useful for publishing state changes without sending redundant updates:

from pydoover.utils import on_change

class DeviceMonitor:
    def status_changed(self, new_value, old_value, is_first, name):
        if is_first:
            print(f"{name} initial value: {new_value}")
        else:
            print(f"{name} changed from {old_value} to {new_value}")

    @on_change("status_changed", name="pump_status")
    def get_pump_status(self):
        return self.read_digital_input(1)

The callback receives four arguments:

ArgumentDescription
new_resultThe new return value of the decorated function.
old_resultThe previous return value, or None on the first call.
is_firstTrue if this is the first time the function has returned a value.
change_detector_nameThe name parameter, or the function name if name was not provided.

The callback can be specified as either a string (the name of a method on self) or a direct callable reference. State is tracked per-instance, so multiple objects of the same class maintain independent change detection.

Sensor Mapping

map_reading

Maps a raw sensor reading (typically from a 4-20mA current loop) to an output value range:

from pydoover.utils import map_reading

# Map a 4-20mA reading to 0-100 degrees
temperature = map_reading(12.0, output_values=[0, 100])
# Result: 50.0 (12mA is midpoint of 4-20mA range)

# Values below 3mA return None (sensor disconnected)
result = map_reading(2.5, output_values=[0, 100])
# Result: None
ParameterTypeDefaultDescription
in_valfloat--The raw sensor reading.
output_valueslist[float]--The output range to map to. Must have the same length as raw_readings.
raw_readingslist[float][4, 20]The raw input range breakpoints.
ignore_belowfloat3Values below this threshold return None (sensor fault detection).

The function supports multi-point calibration by providing more than two values in both raw_readings and output_values:

# Three-point calibration for a non-linear sensor
level = map_reading(
    in_val=10.0,
    output_values=[0, 40, 100],    # Output breakpoints
    raw_readings=[4, 12, 20],      # Corresponding raw readings
)

Other Utilities

sanitize_display_name

Normalises a string for use as an identifier by replacing spaces with underscores, removing special characters, and converting to lowercase:

from pydoover.utils.utils import sanitize_display_name

name = sanitize_display_name("My Sensor (v2)")
# Result: "my_sensor_v2"

Dictionary Search

Two functions for searching nested dictionaries:

from pydoover.utils import find_object_with_key, find_path_to_key

data = {"a": {"b": {"target": 42}}}

# Find the value for a key anywhere in the structure
value = find_object_with_key(data, "target")
# Result: 42

# Find the dot-separated path to a key
path = find_path_to_key(data, "target")
# Result: "a.b.target"

CaseInsensitiveDict

A dictionary that normalises keys to lowercase for case-insensitive lookups:

from pydoover.utils import CaseInsensitiveDict

d = CaseInsensitiveDict({"Temperature": 22.5})
print(d["temperature"])  # 22.5
print(d["TEMPERATURE"])  # 22.5

setup_logging

Configures the root logger with coloured console output:

from pydoover.utils import setup_logging

setup_logging(debug=True)  # DEBUG level with coloured formatter

You can pass a custom formatter and a list of filters for more control.

Related Pages