vehicle_common.runtime.managed_registry

Attributes

ManagedRegistryKind

Classes

ManagedCommDescriptor
ManagedCommSpec
ManagedCommRegistry

Module Contents

ManagedRegistryKind = ('publisher', 'subscription', 'client')

class ManagedCommDescriptor

kind: str

scope: vehicle_common.runtime.comm_naming.ManagedCommScope

lifetime: vehicle_common.runtime.comm_policy.ManagedCommLifetime

owner_label: str

interface_name: str

name: str

peers: tuple[str, ...]

class ManagedCommSpec

kind: str

scope: vehicle_common.runtime.comm_naming.ManagedCommScope

lifetime: vehicle_common.runtime.comm_policy.ManagedCommLifetime

owner: object

owner_label: str

interface_type: object

name: str

args: tuple[Any, ...]

kwargs: dict[str, Any]

peers: tuple[str, ...]

descriptor() → ManagedCommDescriptor

class ManagedCommRegistry(*, connection_status: Callable[[], Mapping[str, bool]], raw_create_publisher: Callable[..., object], raw_create_subscription: Callable[..., object], raw_create_client: Callable[..., object], raw_destroy_publisher: Callable[[object], bool | None], raw_destroy_subscription: Callable[[object], bool | None], raw_destroy_client: Callable[[object], bool | None])

_connection_status

_raw_create_publisher

_raw_create_subscription

_raw_create_client

_raw_destroy_publisher

_raw_destroy_subscription

_raw_destroy_client

_entities: dict[int, vehicle_common.runtime.managed_entities.ManagedEntity]

debug_descriptors() → tuple[ManagedCommDescriptor, ...]

register(*, kind: str, scope: vehicle_common.runtime.comm_naming.ManagedCommScope, lifetime: vehicle_common.runtime.comm_policy.ManagedCommLifetime, owner: object, owner_label: str, interface_type: object, name: str, args: tuple[Any, ...], kwargs: dict[str, Any], peers: tuple[str, ...])

destroy_for_owner(owner: object, *, lifetime: vehicle_common.runtime.comm_policy.ManagedCommLifetime | None = None) → None

destroy_publisher(publisher) → bool

destroy_subscription(subscription) → bool

destroy_client(client) → bool

on_peer_connection_change(peer_name: str, *, connected: bool) → None

close() → None

_make_wrapper(spec: ManagedCommSpec) → vehicle_common.runtime.managed_entities.ManagedEntity

_create_managed_underlying(wrapper: vehicle_common.runtime.managed_entities.ManagedEntity)

_destroy_managed_underlying(wrapper: vehicle_common.runtime.managed_entities.ManagedEntity) → None

_should_have_managed_underlying(wrapper: vehicle_common.runtime.managed_entities.ManagedEntity) → bool

_ensure_managed_underlying(wrapper: vehicle_common.runtime.managed_entities.ManagedEntity) → None

_refresh_managed_underlying(wrapper: vehicle_common.runtime.managed_entities.ManagedEntity) → None

_destroy_managed_entity(wrapper: vehicle_common.runtime.managed_entities.ManagedEntity) → bool