caret_analyze.architecture.struct package

Submodules

Module contents

class caret_analyze.architecture.struct.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.CallbackGroupStruct(callback_group_type: CallbackGroupType, node_name: str, callbacks: list[CallbackStruct], callback_group_name: str)

Bases: object

Callback group value object.

property callback_group_name: str

property callback_group_type: CallbackGroupType: Get callback_group_type.

Returns

CallbackGroupType

property callback_group_type_name: str: Get callback_group_type name.

Returns

CallbackGroupType name

property callback_names: list[str]

property callbacks: list[CallbackStruct]

property node_name: str

Get node name.

Returns

str: node name

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

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

to_value() → CallbackGroupStructValue

class caret_analyze.architecture.struct.CallbackStruct(node_name: str, symbol: str, subscribe_topic_name: str | None, service_name: str | None, publish_topics: list[PublishTopicInfoValue] | None, construction_order: int, callback_name: str)

Bases: object

Callback value base class.

property callback_name: str

Get callback name.

Returns

str: callback name

abstract property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: callback type

property callback_type_name: str

property construction_order: int

Get construction order.

Returns

int: construction order

insert_publisher(publish_topic_info: PublishTopicInfoValue) → None

property node_name: str

Get node name.

Returns

str: node name

property publish_topics: list[PublishTopicInfoValue] | None

Get publisher topic info.

Returns

PublishTopicInfo: publish topics

remove_publisher(publish_topic_info: PublishTopicInfoValue) → None

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

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

property service_name: str | None

property subscribe_topic_name: str | None

property symbol: str

Get callback symbol name.

Returns

str: callback symbol name

abstract to_value() → CallbackStructValue

class caret_analyze.architecture.struct.CommunicationStruct(node_publish: NodeStruct, node_subscription: NodeStruct, publisher_value: PublisherStruct, subscription_value: SubscriptionStruct, publish_callback_values: list[CallbackStruct] | None, subscription_callback_value: CallbackStruct | None)

Bases: object

property publish_callback_names: list[str] | None

property publish_callbacks: list[CallbackStruct] | None

property publish_node: NodeStruct

property publish_node_name: str

property publisher: PublisherStruct

property publisher_construction_order: int | None

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

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

property subscribe_callback: CallbackStruct | None

property subscribe_callback_name: str | None

property subscribe_node: NodeStruct

property subscribe_node_name: str

property subscription: SubscriptionStruct

property subscription_construction_order: int | None

to_value() → CommunicationStructValue

property topic_name: str

class caret_analyze.architecture.struct.ExecutorStruct(executor_type: ExecutorType, callback_groups: list[CallbackGroupStruct], executor_name: str)

Bases: object

Executor info for architecture.

property callback_group_names: list[str]

property callback_groups: list[CallbackGroupStruct]

property callback_names: list[str]

property callbacks: list[CallbackStruct]

property executor_name: str

property executor_type: ExecutorType

property executor_type_name: str

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

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

to_value() → ExecutorStructValue

class caret_analyze.architecture.struct.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.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.NodePathStruct(node_name: str, subscription: SubscriptionStruct | None, publisher: PublisherStruct | None, child: list[CallbackStruct | VariablePassingStruct] | None, message_context: MessageContextStruct | None)

Bases: object

property callback_names: list[str] | None

property callbacks: list[CallbackStruct] | None

property child: list[CallbackStruct | VariablePassingStruct] | None

property message_context: MessageContextStruct | None

property message_context_type: MessageContextType | None

property node_name: str

property publish_topic_name: str | None

property publisher: PublisherStruct | None

property publisher_construction_order: int | None

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

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

property subscribe_topic_name: str | None

property subscription: SubscriptionStruct | None

property subscription_callback: SubscriptionCallbackStruct | None

property subscription_construction_order: int | None

to_value() → NodePathStructValue

property variable_passings: list[VariablePassingStruct] | None

class caret_analyze.architecture.struct.NodeStruct(node_name: str, publishers: list[PublisherStruct], subscriptions_info: list[SubscriptionStruct], services: list[ServiceStruct], timers: list[TimerStruct], node_paths: list[NodePathStruct], callback_groups: list[CallbackGroupStruct] | None, variable_passings: list[VariablePassingStruct] | None)

Bases: object

Executor info for architecture.

property callback_group_names: list[str] | None

property callback_groups: list[CallbackGroupStruct] | None

property callback_names: list[str] | None

property callbacks: list[CallbackStruct] | None

get_publisher(publish_topic_name: str, construction_order: int | None) → PublisherStruct

get_publisher_from_callback(callback_name: str) → list[PublisherStruct]

get_service(service_name: str, construction_order: int | None) → ServiceStruct

get_subscription(subscribe_topic_name: str, construction_order: int | None = None) → SubscriptionStruct

get_subscription_from_callback(callback_name: str) → SubscriptionStruct | None

insert_publisher_callback(publish_topic_name: str, callback_name: str, publisher_construction_order: int) → None

insert_variable_passing(callback_name_write: str, callback_name_read: str) → None

property node_name: str

property paths: list[NodePathStruct]

property publish_topic_names: list[str]

property publishers: list[PublisherStruct]

remove_publisher_and_callback(publish_topic_name: str, callback_name: str, publisher_construction_order: int) → None

remove_variable_passing(callback_name_write: str, callback_name_read: str) → None

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

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

property services: list[ServiceStruct]

property subscribe_topic_names: list[str]

property subscriptions: list[SubscriptionStruct]

property timers: list[TimerStruct]

to_value() → NodeStructValue

update_node_path(paths: list[NodePathStruct]) → None

property variable_passings: list[VariablePassingStruct] | None

class caret_analyze.architecture.struct.PathStruct(path_name: str | None, child: list[NodePathStruct | CommunicationStruct])

Bases: object

property child: list[NodePathStruct | CommunicationStruct]

property child_names: list[str]

property communications: list[CommunicationStruct]

property node_names: list[str]

property node_paths: list[NodePathStruct]

property path_name: str | None

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

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

to_value() → PathStructValue

property topic_names: list[str]

class caret_analyze.architecture.struct.PublisherStruct(node_name: str, topic_name: str, callback_values: list[CallbackStruct] | None, construction_order: int)

Bases: object

Structured publisher value.

property callback_names: list[str] | None

property callbacks: list[CallbackStruct] | None

property construction_order: int

insert_callback(callback: CallbackStruct) → None

property node_name: str

remove_callback(callback: CallbackStruct) → None

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

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

to_value() → PublisherStructValue

property topic_name: str

class caret_analyze.architecture.struct.ServiceCallbackStruct(node_name: str, symbol: str, service_name: str, publish_topics: list[PublishTopicInfoValue] | None, construction_order: int, callback_name: str)

Bases: CallbackStruct

Structured service callback value.

property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: callback type

property service_name: str

to_value() → ServiceCallbackStructValue

class caret_analyze.architecture.struct.ServiceStruct(node_name: str, service_name: str, callback_info: ServiceCallbackStruct | None, construction_order: int)

Bases: object

Service info.

property callback: ServiceCallbackStruct | None

property callback_name: str | None

property construction_order: int

property node_name: str

property service_name: str

to_value() → ServiceStructValue

class caret_analyze.architecture.struct.SubscriptionCallbackStruct(node_name: str, symbol: str, subscribe_topic_name: str, publish_topics: list[PublishTopicInfoValue] | None, construction_order: int, callback_name: str)

Bases: CallbackStruct

Structured subscription callback value.

property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: callback type

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

property subscribe_topic_name: str

to_value() → SubscriptionCallbackStructValue

class caret_analyze.architecture.struct.SubscriptionStruct(node_name: str, topic_name: str, callback_info: SubscriptionCallbackStruct | None, construction_order: int)

Bases: object

Subscription info.

property callback: SubscriptionCallbackStruct | None

property callback_name: str | None

property construction_order: int

property node_name: str

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

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

to_value() → SubscriptionStructValue

property topic_name: str

class caret_analyze.architecture.struct.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.TimerCallbackStruct(node_name: str, symbol: str, period_ns: int, publish_topics: list[PublishTopicInfoValue] | None, construction_order: int, callback_name: str)

Bases: CallbackStruct

Structured timer callback value.

property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: callback type

property period_ns: int

to_value() → TimerCallbackStructValue

class caret_analyze.architecture.struct.TimerStruct(node_name: str, period_ns: int, callback_info: TimerCallbackStruct | None, construction_order: int)

Bases: object

Timer info.

property callback: TimerCallbackStruct | None

property callback_name: str | None

property construction_order: int

property node_name: str

property period_ns: int

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

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

to_value() → TimerStructValue

class caret_analyze.architecture.struct.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

class caret_analyze.architecture.struct.VariablePassingStruct(node_name: str, callback_write: CallbackStruct, callback_read: CallbackStruct)

Bases: object

variable passing info.

property callback_name_read

Get read-side callback name.

Returns

[str]: read-side callback name.

property callback_name_write: str

Get write-side callback name.

Returns

[str]: write-side callback name.

property callback_read: CallbackStruct

property callback_write: CallbackStruct

property node_name: str

Get node name.

Returns

str: node name

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

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

to_value() → VariablePassingStructValue