Source code for labgrid.binding

import enum
from functools import wraps
from typing import Any, Dict

import attr

[docs]@attr.s(eq=False) class StateError(Exception): msg = attr.ib(validator=attr.validators.instance_of(str))
[docs]@attr.s(eq=False) class BindingError(Exception): msg = attr.ib(validator=attr.validators.instance_of(str))
[docs]@enum.unique class BindingState(enum.Enum): error = -1 idle = 0 bound = 1 active = 2
[docs]@attr.s(eq=False) class BindingMixin: """ Handles the binding and activation of drivers and their supplying resources and drivers. One client can be bound to many suppliers, and one supplier can be bound by many clients. Conflicting access to one supplier can be avoided by deactivating conflicting clients before activation (using the resolve_conflicts callback). """ bindings: Dict[str, Any] = {} # these are controlled by the Target target = attr.ib() name = attr.ib( validator=attr.validators.optional(attr.validators.instance_of(str))) state = attr.ib(default=BindingState.idle, init=False)
[docs] def __attrs_post_init__(self): self.suppliers = set() self.clients = set() target = if target is not None: # bind will set it again if successful = None target.bind(self) assert is not None
@property def display_name(self): if return f"{self.__class__.__name__}(target={}, name={})" return f"{self.__class__.__name__}(target={})"
[docs] def on_supplier_bound(self, supplier): """Called by the Target after a new supplier has been bound""" pass
[docs] def on_client_bound(self, client): """Called by the Target after a new client has been bound""" pass
[docs] def on_activate(self): """Called by the Target when this object has been activated""" pass
[docs] def on_deactivate(self): """Called by the Target when this object has been deactivated""" pass
[docs] def resolve_conflicts(self, client): """ Called by the Target to allow this object to deactivate conflicting clients. """ pass
[docs] @classmethod def check_active(cls, func): @wraps(func) def wrapper(self, *_args, **_kwargs): if self.state is not raise StateError( f'{self} has not been activated, {func.__qualname__} cannot be called in state "{}"' # pylint: disable=line-too-long ) return func(self, *_args, **_kwargs) return wrapper
[docs] class NamedBinding: """ Marks a binding (or binding set) as requiring an explicit name. """
[docs] def __init__(self, value): self.value = value
[docs] def __repr__(self): return f"Binding.Named({repr(self.value)})"