Synchronization

Synchronizer(threshold)

Synchronizer. Syncs elements from multiple sources according to a threshold in absolute difference.

Source code in src/compas_camcal/util/sync.py

def __init__(self, threshold: float):
    self._threshold = threshold

sync(*collections, key=None)

Synchronize values from multiple sources according to a threshold in absolute difference. Optionally, a key function can be provided to extract the value to compare from each element.

Parameters:

Name	Type	Description	Default
collections	Tuple[Any]	The collections to synchronize.	()
key	Optional[Callable[[Any], float]]	The function to extract the value to compare from each element. If None, the elements themselves are compared.	None

Yields:

Type	Description
Iterable[Any]	Synchronized sets of elements.

Examples:

Sync floats by value

>>> list(Synchronizer(0.1).sync([0.96, 2.01, 2.97], [1.18, 1.99, 3.02, 4.03]))
[[2.01, 1.99], [2.97, 3.02]]

Sync characters by ordinal value (key function)

>>> list(Synchronizer(5).sync(["a", "e", "l"], ["b", "d", "f", "h"], key=ord))
[['a', 'b'], ['e', 'd'], ['l', 'h']]

Source code in src/compas_camcal/util/sync.py

def sync(
    self, *collections: Tuple[Any], key: Optional[Callable[[Any], float]] = None
) -> Iterable[Any]:
    """
    Synchronize values from multiple sources according to a threshold in absolute difference.
    Optionally, a key function can be provided to extract the value to compare from each element.

    Args:
        collections: The collections to synchronize.
        key: The function to extract the value to compare from each element. If None, the elements themselves are compared.

    Yields:
        Synchronized sets of elements.

    Examples:
        # Sync floats by value
        >>> list(Synchronizer(0.1).sync([0.96, 2.01, 2.97], [1.18, 1.99, 3.02, 4.03]))
        [[2.01, 1.99], [2.97, 3.02]]

        # Sync characters by ordinal value (key function)
        >>> list(Synchronizer(5).sync(["a", "e", "l"], ["b", "d", "f", "h"], key=ord))
        [['a', 'b'], ['e', 'd'], ['l', 'h']]
    """
    # Create a sorted list for each collection ("queue")
    queues = [sorted(collection, key=key) for collection in collections]

    # Loop until any queue is empty
    while all(queues):
        # Get the front of each queue
        fronts = [queue[0] for queue in queues]

        # Get the minimum and maximum key values
        min_front = min(fronts, key=key)
        max_front = max(fronts, key=key)

        if key is not None:  # If a key function was provided, apply it
            min_value = key(min_front)
            max_value = key(max_front)
        else:  # Otherwise, use the elements themselves
            min_value = min_front
            max_value = max_front

        # Get the absolute difference
        difference = abs(max_value - min_value)

        # If the difference is within the threshold, yield the synchronized elements
        if difference <= self._threshold:
            yield fronts
            for queue in queues:  # Pop the front from each queue
                queue.pop(0)
        else:  # Otherwise, pop the minimum value from its queue
            min_queue = queues[fronts.index(min_front)]
            min_queue.pop(0)