caret_analyze.architecture.struct.message_context module

class caret_analyze.architecture.struct.message_context.CallbackChainStruct(node_name: str, message_context_dict: dict, subscription: SubscriptionStruct | None, publisher: PublisherStruct | None, callbacks: list[CallbackStruct] | None)

Bases: MessageContextStruct

TYPE_NAME = 'callback_chain'

Callback chain.

Latency is calculated from callback durations in the node path. When a path within a node passes through multiple callbacks, it is assumed that messages are passed between callbacks by a buffer of queue size 1 (ex. a member variable that stores a single message). If the queue size is larger than 1, the node latency may be calculated to be small.

property context_type: MessageContextType

is_applicable_path(subscription: SubscriptionStruct | None, publisher: PublisherStruct | None, callbacks: list[CallbackStruct] | None) → bool

to_dict() → dict

to_value() → CallbackChain

class caret_analyze.architecture.struct.message_context.InheritUniqueStampStruct(node_name: str, message_context_dict: dict, subscription: SubscriptionStruct | None, publisher: PublisherStruct | None, child: list[CallbackStruct] | None)

Bases: MessageContextStruct

TYPE_NAME = 'inherit_unique_stamp'

Inherit header timestamp.

Latency is calculated for pub/sub messages with the same timestamp value. If the input timestamp is not unique, it may calculate an incorrect value.

property context_type: MessageContextType

to_value() → InheritUniqueStamp

class caret_analyze.architecture.struct.message_context.MessageContextStruct(node_name: str, message_context_dict: dict, subscription: SubscriptionStruct | None, publisher: PublisherStruct | None, child: list[CallbackStruct] | None)

Bases: object

Structured message context value.

property callbacks: list[CallbackStruct] | None

abstract property context_type: MessageContextType

static create_instance(context_type_name: str, context_dict: dict, node_name: str, subscription: SubscriptionStruct | None, publisher: PublisherStruct | None, child: list[CallbackStruct] | None) → MessageContextStruct

is_applicable_path(subscription: SubscriptionStruct | None, publisher: PublisherStruct | None, callbacks: list[CallbackStruct] | None) → bool

property node_name: str

property publisher_construction_order: int | None

property publisher_topic_name: str | None

rename_node(src: str, dst: str) → None

rename_topic(src: str, dst: str) → None

property subscription_construction_order: int | None

property subscription_topic_name: str | None

to_dict() → dict

abstract to_value() → MessageContext

property type_name: str

class caret_analyze.architecture.struct.message_context.TildeStruct(node_name: str, message_context_dict: dict, subscription: SubscriptionStruct | None, publisher: PublisherStruct | None, callbacks: list[CallbackStruct] | None)

Bases: MessageContextStruct

TYPE_NAME = 'tilde'

tilde.

Latency is calculated from tilde.

property context_type: MessageContextType

is_applicable_path(subscription: SubscriptionStruct | None, publisher: PublisherStruct | None, callbacks: list[CallbackStruct] | None) → bool

to_value() → Tilde

class caret_analyze.architecture.struct.message_context.UseLatestMessageStruct(node_name: str, message_context_dict: dict, subscription: SubscriptionStruct | None, publisher: PublisherStruct | None, child: list[CallbackStruct] | None)

Bases: MessageContextStruct

TYPE_NAME = 'use_latest_message'

Use message context

property context_type: MessageContextType

to_value() → UseLatestMessage