Source code for statetracker.ops.synced_to_op

"""synced_to - Create or pair synced states."""

from ..imports import Integral, Optional, State_type, beartype


[docs] @beartype def synced_to( child_state: State_type, name: Optional[str] = None, num_values: Optional[Integral] = None ) -> State_type: """Create a new state synchronized with child_state. Args: child_state: The state to synchronize with. name: Optional name for the new state. num_values: Optional number of values for the new state. If not provided, uses child_state.num_values. Can be different from child_state.num_values - the group will track the max and states receive None when out of range. Returns: A new State synchronized with child_state. """ from ..state import State nv = num_values if num_values is not None else child_state.num_values synced = State(num_values=nv, name=name) synced.sync_with(child_state) # Set iter_order to minimum in the sync group (ensures earliest iteration priority) min_iter_order = min(s.iter_order for s in synced._synced_group._states) synced.iter_order = min_iter_order return synced
[docs] @beartype def sync(a: State_type, b: State_type) -> None: """Synchronize two existing states (bidirectional).""" a.sync_with(b)