caret_analyze.value_objects package

Submodules

Module contents

Package defining the value object class.

Defined classes are used as an interface across multiple packages: architecture, infra, and runtime. CARET uses value objects to avoid unexpectedly overwritten.

class caret_analyze.value_objects.CallbackChain(node_name: str, message_context_dict: dict, subscription: SubscriptionStructValue | None, publisher: PublisherStructValue | None, callbacks: tuple[CallbackStructValue, ...] | None)

Bases: MessageContext

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

Get context type.

Returns

MessageContextType: Message context type.

is_applicable_path(subscription: SubscriptionStructValue | None, publisher: PublisherStructValue | None, callbacks: tuple[CallbackStructValue, ...] | None) → bool

Get applicable path.

Parameters

subscriptionSubscriptionStructValue | None: Target subscription value.
publisherPublisherStructValue | None: Target publisher value.
callbackstuple[CallbackStructValue, …] | None: Target callbacks.

Returns

bool: True if applicable path, false otherwise.

to_dict() → dict

Get to dict.

Returns

dict: Dict.

verify() → bool

Get verify.

Returns

bool: Verify or not.

class caret_analyze.value_objects.CallbackGroupStructValue(callback_group_type: CallbackGroupType, node_name: str, callback_values: tuple[CallbackStructValue, ...], callback_group_name: str)

Bases: ValueObject, Summarizable

Callback group value object.

property callback_group_name: str

Get callback group name.

Returns

str: Callback group name.

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: tuple[str, ...]

Get callback names.

Returns

tuple[str, …]: Callback names which are added to the callback group.

property callbacks: tuple[CallbackStructValue, ...]

Get callbacks.

Returns

tuple[CallbackStructValue, …]: Callbacks which are added to the callback group.

property node_name: str

Get node name.

Returns

str: node name

property summary: Summary

Get summary.

Returns

Summary: summary info.

class caret_analyze.value_objects.CallbackGroupType(name: str)

Bases: ValueObject

Callback group type class.

The types of callback groups are as follows:

MUTUALLY_EXCLUSIVE
REENTRANT

MUTUALLY_EXCLUSIVE: CallbackGroupType = <caret_analyze.value_objects.callback_group.CallbackGroupType object>

REENTRANT: CallbackGroupType = <caret_analyze.value_objects.callback_group.CallbackGroupType object>

property type_name: str

Return callback group type name.

Returns

str: type name.

class caret_analyze.value_objects.CallbackGroupValue(callback_group_type_name: str, node_name: str, node_id: str, callback_ids: tuple[str, ...], callback_group_id: str, *, callback_group_name: str | None = None)

Bases: ValueObject

Callback group value object.

property callback_group_id: str

Get callback group id.

Returns

str: Callback group id.

property callback_group_name: str | None

Get callback group name.

Returns

str | None: Callback group name.

property callback_group_type: CallbackGroupType: Get callback_group_type.

Returns

CallbackGroupType

property callback_ids: tuple[str, ...]

Get callback ids.

Returns

tuple[str, …]: callback ids added to the callback group.

property node_id: str

Get node id.

Returns

str: Node id.

property node_name: str

Get node name.

Returns

str: Node name which includes the callback group.

class caret_analyze.value_objects.CallbackStructValue(node_name: str, symbol: str, subscribe_topic_name: str | None, service_name: str | None, publish_topics: tuple[PublishTopicInfoValue, ...] | None, construction_order: int, callback_name: str)

Bases: Summarizable

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

Get callback type name.

Returns

str: Callback type name.

property construction_order: int

Get construction order.

Returns

int: Construction order.

property node_name: str

Get node name.

Returns

str: Node name.

property publish_topics: tuple[PublishTopicInfoValue, ...] | None

Get publisher topic info.

Returns

tuple[PublishTopicInfoValue, …] | None: publish topics

property service_name: str | None

Get service name.

Returns

str | None: Service name.

property subscribe_topic_name: str | None

Get subscribe topic name.

Returns

str | None: Subscribe topic name.

abstract property summary: Summary

Get summary.

Returns

Summary: Summary about value objects and runtime data objects.

property symbol: str

Get callback symbol name.

Returns

str: Callback symbol name.

class caret_analyze.value_objects.CallbackType(name: str)

Bases: ValueObject

callback group type class.

SERVICE: CallbackType = <caret_analyze.value_objects.callback.CallbackType object>

SUBSCRIPTION: CallbackType = <caret_analyze.value_objects.callback.CallbackType object>

TIMER: CallbackType = <caret_analyze.value_objects.callback.CallbackType object>

property type_name: str

Return callback type name.

Returns

str: Type name.

class caret_analyze.value_objects.CallbackValue(callback_id: str, node_name: str, node_id: str, symbol: str, subscribe_topic_name: str | None, service_name: str | None, publish_topics: tuple[PublishTopicInfoValue, ...] | None, construction_order: int, *, callback_name: str | None = None)

Bases: ValueObject

Value object class for representing a callback.

This class has minimal information and no structure. It’s used as the return value of ArchitectureReader.

property callback_id: str

Get callback id.

Callback id is is used to bind. Callback id should be the same if the node name and other properties are the same. If any properties is different, it should be a different callback id.

Returns

str: Callback unique id.

property callback_name: str | None

Get callback name.

Returns

str: Callback name.

Note:

Different architecture_readers may return different values.

abstract property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: Callback type.

property construction_order: int

Get construction order.

Returns

int: Construction order.

property node_id: str

Get node id.

Returns

str: Node id.

property node_name: str

Get node name.

Returns

str: Node name.

property publish_topics: tuple[PublishTopicInfoValue, ...] | None

Get publisher topic info.

Returns

tuple[PublishTopicInfoValue, …]] | None: publish topics

Note:

Since callback publishes multiple topics, there are multiple publish topic names & construction_order.

property service_name: str | None

Get service name.

Returns

str: Service name.

property subscribe_topic_name: str | None

Get subscription topic name.

Returns

str | None: Topic name which the callback subscribes.

Note:

Only one subscription callback have a single subscribe topic name.

property symbol: str

Get callback symbol name.

Returns

str: Callback symbol name.

class caret_analyze.value_objects.CommunicationStructValue(node_publish: NodeStructValue, node_subscription: NodeStructValue, publisher_value: PublisherStructValue, subscription_value: SubscriptionStructValue, publish_callback_values: tuple[CallbackStructValue, ...] | None, subscription_callback_value: CallbackStructValue | None)

Bases: ValueObject, Summarizable

property publish_callback_names: tuple[str, ...] | None

property publish_callbacks: tuple[CallbackStructValue, ...] | None

property publish_node: NodeStructValue

property publish_node_name: str

property publisher: PublisherStructValue

property publisher_construction_order: int

property subscribe_callback: CallbackStructValue | None

property subscribe_callback_name: str | None

property subscribe_node: NodeStructValue

property subscribe_node_name: str

property subscription: SubscriptionStructValue

property subscription_construction_order: int

property summary: Summary

Get summary.

Returns

Summary: summary info.

property topic_name: str

class caret_analyze.value_objects.DiffNode(left_node: NodeStructValue, right_node: NodeStructValue)

Bases: object

diff_node_pubs() → tuple[tuple[str, ...], tuple[str, ...]]

Compare two nodes of architecture objects and return the difference of publish topic names.

Returns

tuple[tuple[str, …], tuple[str, …]]: Returns publish topic names that exist only in the respective nodes.

diff_node_subs() → tuple[tuple[str, ...], tuple[str, ...]]

Compare two nodes of architecture objects and return the difference of subscribe topic names.

Returns

tuple[tuple[str, …], tuple[str, …]]: Returns subscribe topic names that exist only in the respective nodes.

class caret_analyze.value_objects.ExecutorStructValue(executor_type: ExecutorType, callback_group_values: tuple[CallbackGroupStructValue, ...], executor_name: str)

Bases: ValueObject, Summarizable

Executor info for architecture.

property callback_group_names: tuple[str, ...]

Get callback group names.

Returns

tuple[str, …]: Callback group names.

property callback_groups: tuple[CallbackGroupStructValue, ...]

Get callback groups.

Returns

tuple[CallbackGroupStructValue, …]: Construct callback group value object.

property callback_names: tuple[str, ...]

Get callback name list.

Returns

tuple[str, …]: Callback names.

property callbacks: tuple[CallbackStructValue, ...]

Get Callbacks.

Returns

tuple[CallbackStructValue, …]: Callback value base.

property executor_name: str

Get executor name.

Returns

str: Executor name.

property executor_type: ExecutorType

Get executor type.

Returns

ExecutorType: Executor type.

property executor_type_name: str

Get executor type name.

Returns

str: Executor type name.

property summary: Summary

Get summary.

Returns

Summary: Summary about value objects and runtime data objects.

class caret_analyze.value_objects.ExecutorType(type_name: str)

Bases: ValueObject

executor type class.

MULTI_THREADED_EXECUTOR: ExecutorType = <caret_analyze.value_objects.executor.ExecutorType object>

SINGLE_THREADED_EXECUTOR: ExecutorType = <caret_analyze.value_objects.executor.ExecutorType object>

property type_name: str

Get executor type name.

Returns

str: Type name.

class caret_analyze.value_objects.ExecutorValue(executor_type_name: str, callback_group_ids: tuple[str, ...], *, executor_name: str | None = None)

Bases: ValueObject

Executor value class.

property callback_group_ids: tuple[str, ...]

Get callback group ids.

Returns

tuple[str, …]: Callback group id list.

property executor_name: str | None

Get executor name.

Returns

str | None: Executor name.

property executor_type: ExecutorType

Get executor type.

Returns

ExecutorType: Executor type.

class caret_analyze.value_objects.InheritUniqueStamp(node_name: str, message_context_dict: dict, subscription: SubscriptionStructValue | None, publisher: PublisherStructValue | None, child: tuple[CallbackStructValue, ...] | None)

Bases: MessageContext

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

Get context type.

Returns

MessageContextType: Message context type.

verify() → bool

Get verify.

Returns

bool: Same or difference.

class caret_analyze.value_objects.MessageContext(node_name: str, message_context_dict: dict, subscription: SubscriptionStructValue | None, publisher: PublisherStructValue | None, child: tuple[CallbackStructValue, ...] | None)

Bases: ValueObject, Summarizable

Structured message context value.

property callbacks: tuple[CallbackStructValue, ...] | None

Get callbacks.

Returns

tuple[CallbackStructValue, …] | None: Callback struct values.

abstract property context_type: MessageContextType

Get context type.

Returns

MessageContextType: Message context type.

static create_instance(context_type_name: str, context_dict: dict, node_name: str, subscription: SubscriptionStructValue | None, publisher: PublisherStructValue | None, child: tuple[CallbackStructValue, ...] | None) → MessageContext

Get create instance.

Parameters

context_type_namestr: Context type name.
context_dictdict: Context dict.
node_namestr: Node name.
subscriptionSubscriptionStructValue | None: Target subscription value.
publisher: PublisherStructValue | None: Target publisher.
childtuple[CallbackStructValue, …] | None: Child elements.

Returns

MessageContext: Message context.

is_applicable_path(subscription: SubscriptionStructValue | None, publisher: PublisherStructValue | None, callbacks: tuple[CallbackStructValue, ...] | None) → bool

Get applicable path.

Parameters

subscriptionSubscriptionStructValue | None: Target subscription value.
publisherPublisherStructValue | None: Target publisher value.
callbackstuple[CallbackStructValue, …] | None: Target callbacks.

Returns

bool: True if applicable path, false otherwise.

property node_name: str

Get node name.

Returns

str: Node name.

property publisher_construction_order: int | None

Get publisher topic name.

Returns

int | None: Publisher construction order.

property publisher_topic_name: str | None

Get publisher topic name.

Returns

str | None: Publisher topic name.

property subscription_construction_order: int | None

Get subscription construction order.

Returns

int | None: Subscription construction order.

property subscription_topic_name: str | None

Get subscription topic name.

Returns

str | None: Subscription topic name.

property summary: Summary

Get summary.

Returns

Summary: Summary.

to_dict() → dict

Get to dict.

Returns

dict: Dict.

property type_name: str

Get type name.

Returns

str: Type name.

abstract verify() → bool

Get verify.

Returns

bool: Verify or not.

class caret_analyze.value_objects.MessageContextType(type_name: str)

Bases: ValueObject

Message context type.

CALLBACK_CHAIN: MessageContextType = <caret_analyze.value_objects.message_context.MessageContextType object>

INHERIT_UNIQUE_STAMP: MessageContextType = <caret_analyze.value_objects.message_context.MessageContextType object>

TILDE: MessageContextType = <caret_analyze.value_objects.message_context.MessageContextType object>

USE_LATEST_MESSAGE: MessageContextType = <caret_analyze.value_objects.message_context.MessageContextType object>

property type_name: str

Get type name.

Parameters

str: Type name.

class caret_analyze.value_objects.NodePathStructValue(node_name: str, subscription: SubscriptionStructValue | None, publisher: PublisherStructValue | None, child: tuple[CallbackStructValue | VariablePassingStructValue, ...] | None, message_context: MessageContext | None)

Bases: ValueObject, Summarizable

StructValue object class for representing a node path.

This class is a structure that includes other related StructValue classes, such as callbacks, and used as the return value of Architecture object. In CARET, the node path is defined as from subscribe to publish.

property callback_names: tuple[str, ...] | None

Get callback names.

Returns

tuple[str, …] | None: Callback names of all child elements that comprise the node path.

property callbacks: tuple[CallbackStructValue, ...] | None

Get callbacks.

Returns

tuple[CallbackStructValue, …] | None: Callbacks in all child elements that comprise the node path.

property child: tuple[CallbackStructValue | VariablePassingStructValue, ...] | None

Get child elements.

Returns

tuple[CallbackStructValue | VariablePassingStructValue, …] | None: Child elements for this node path.

property message_context: MessageContext | None

Get message context.

Returns

MessageContext | None: Message context for this node path.

property message_context_type: MessageContextType | None

Get message context type.

Returns

MessageContextType | None: Message context type for this node path.

property node_name: str

Get node name.

Returns

str: Node name.

property publish_topic_name: str | None

Get a topic name to publish.

Returns

str | None: Topic name to publish.

property publisher: PublisherStructValue | None

Get publisher.

Returns

PublisherStructValue | None: Node path publisher.

property publisher_construction_order: int | None

Get a construction order of publisher.

Returns

int | None: Construction order of publisher.

property subscribe_topic_name: str | None

Get a topic name to subscribe to.

Returns

str | None: Topic name to subscribe to.

property subscription: SubscriptionStructValue | None

Get subscription.

Returns

SubscriptionStructValue | None: Subscription to subscribe to.

property subscription_callback: SubscriptionCallbackStructValue | None

Get subscription callback.

Returns

SubscriptionCallbackStructValue | None: Subscription callback for this node path.

property subscription_construction_order: int | None

Get a construction order of subscription.

Returns

int | None: Construction order of subscription.

property summary: Summary

Get summary.

Returns

Summary: Summary about value objects and runtime data objects.

property variable_passings: tuple[VariablePassingStructValue, ...] | None

Get variable passings.

Returns

tuple[VariablePassingStructValue, …] | None: Variable passings in all child elements that comprise the node path.

class caret_analyze.value_objects.NodePathValue(node_name: str, subscribe_topic_name: str | None, publish_topic_name: str | None, publisher_construction_order: int | None, subscription_construction_order: int | None)

Bases: ValueObject

Value object class for representing a node path.

This class has minimal information and no structure, and used as the return value of ArchitectureReader. In CARET, the node path is defined as from subscribe to publish.

property node_name: str

Get node name.

Returns

str: Node name.

property publish_topic_name: str | None

Get a topic name to publish.

Returns

str | None: Topic name to publish.

property publisher_construction_order: int | None

Get a construction order of publisher.

Returns

int | None: Construction order of publisher.

property subscribe_topic_name: str | None

Get a topic name to subscribe to.

Returns

str | None: Topic name to subscribe to.

property subscription_construction_order: int | None

Get a construction order of subscription.

Returns

int | None: Construction order of subscription.

class caret_analyze.value_objects.NodeStructValue(node_name: str, publishers: tuple[PublisherStructValue, ...], subscriptions_info: tuple[SubscriptionStructValue, ...], services_info: tuple[ServiceStructValue, ...], timers: tuple[TimerStructValue, ...], node_paths: tuple[NodePathStructValue, ...], callback_groups: tuple[CallbackGroupStructValue, ...] | None, variable_passings: tuple[VariablePassingStructValue, ...] | None)

Bases: ValueObject, Summarizable

StructValue object class for representing a node.

This class is a structure that includes other related StructValue classes, such as callbacks, and used as the return value of Architecture object.

property callback_group_names: tuple[str, ...] | None

Get callback group names.

Returns

tuple[str, …] | None: Callback group names that the node contains.

property callback_groups: tuple[CallbackGroupStructValue, ...] | None

Get callback groups.

Returns

tuple[CallbackGroupStructValue, …] | None: Callback groups that the node contains.

property callback_names: tuple[str, ...] | None

Get callback names.

Returns

tuple[str, …] | None: Callback names that the node contains.

property callbacks: tuple[CallbackStructValue, ...] | None

Get callbacks.

Returns

tuple[CallbackStructValue, …] | None: Callbacks that the node contains.

get_path(subscribe_topic_name: str | None, subscription_construction_order: int | None, publish_topic_name: str | None, publisher_construction_order: int | None) → NodePathStructValue

Get node path.

Parameters

subscribe_topic_namestr | None: Topic name to which the node subscribes.
subscription_construction_orderint | None: Construction order to which the node subscribes.
publish_topic_namestr | None: Topic name to which the node publishes.
publisher_construction_orderint | None: Construction order to which the node publishes.

Returns

NodePathStructValue: Node path that matches the condition.

Raises

ItemNotFoundError: Occurs when no items were found.
MultipleItemFoundError: Occurs when several items were found.

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

Get publisher.

Parameters

publish_topic_namestr: Publisher topic name to get.
construction_orderint | None: Construction order to get.

Returns

PublisherStructValue: A publisher that matches the condition.

Raises

ItemNotFoundError: Occurs when no items were found.

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

Get service.

Parameters

service_namestr: Service name to get.
construction_orderint | None: Construction order to get.

Returns

ServiceStructValue: Service instance that matches the condition.

Raises

ItemNotFoundError: Occurs when no items were found.

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

Get subscription.

Parameters

subscribe_topic_namestr: Topic name to get.
construction_orderint | None: Construction order to get.

Returns

SubscriptionStructValue: Subscription instance that matches the condition.

Raises

ItemNotFoundError: Occurs when no items were found.

get_timer(timer_period: int, construction_order: int | None) → TimerStructValue

Get timer.

Parameters

timer_period: int: Timer period to get.
construction_orderint | None: Construction order to get.

Returns

TimerStructValue: A timer that matches the condition.

Raises

ItemNotFoundError: Occurs when no items were found.

property node_name: str

Get node name.

Returns

str: Node name.

property paths: tuple[NodePathStructValue, ...]

Get node paths.

Node paths are defined by subscription and publisher pair.

Returns

tuple[NodePathStructValue, …]: Node paths that the node contains.

property publish_topic_names: tuple[str, ...]

Get topic names to publish.

Returns

tuple[str, …]: Topic names to publish.

property publishers: tuple[PublisherStructValue, ...]

Get publishers.

Returns

tuple[PublisherStructValue, …]: Node path publishers.

property service_names: tuple[str, ...]

Get service names.

Returns

tuple[str, …]: Service names that the node contains.

property services: tuple[ServiceStructValue, ...]

Get services.

Returns

tuple[ServiceStructValue, …]: Services that the node contains.

property subscribe_topic_names: tuple[str, ...]

Get topic names to subscribe to.

Returns

tuple[str, …]: Topic names to subscribe to.

property subscriptions: tuple[SubscriptionStructValue, ...]

Get subscriptions.

Returns

tuple[SubscriptionStructValue, …]: Subscriptions to subscribe to.

property summary: Summary

Get summary.

Returns

Summary: Summary about value objects and runtime data objects.

property timers: tuple[TimerStructValue, ...]

Get timers.

Returns

tuple[TimerStructValue, …]: Timers that the node contains.

property variable_passings: tuple[VariablePassingStructValue, ...] | None

Get variable passings.

Returns

tuple[VariablePassingStructValue, …] | None: Variable passings in all child elements that the node contains.

class caret_analyze.value_objects.NodeValue(node_name: str, node_id: str | None)

Bases: ValueObject

Value object class for representing a node.

This class has minimal information and no structure, and used as the return value of ArchitectureReader.

property node_id: str | None

Get node id.

Returns

str | None: Node id.

property node_name: str

Get node name.

Returns

str: Node name.

class caret_analyze.value_objects.NodeValueWithId(node_name: str, node_id: str)

Bases: NodeValue

Value object class for representing a node path.

This class has minimal information and no structure, and used as the return value of ArchitectureReader.

property node_id: str

Get node id.

Returns

str: Node id.

class caret_analyze.value_objects.PathStructValue(path_name: str | None, child: tuple[NodePathStructValue | CommunicationStructValue, ...])

Bases: ValueObject, Summarizable

property child: tuple[NodePathStructValue | CommunicationStructValue, ...]

property child_names: tuple[str, ...]

property communications: tuple[CommunicationStructValue, ...]

property node_names: tuple[str, ...]

property node_paths: tuple[NodePathStructValue, ...]

property path_name: str | None

property summary: Summary

Get summary.

Returns

Summary: summary info.

property topic_names: tuple[str, ...]

verify() → bool

class caret_analyze.value_objects.PathValue(alias: str, node_path_values: tuple[NodePathValue, ...])

Bases: ValueObject

Path name alias.

property node_path_values: tuple[NodePathValue, ...]

Get node path value.

Returns

tuple[NodePathValue, …]: Node path values.

property path_name: str

Get path name.

Returns

str: Path name.

class caret_analyze.value_objects.PublishTopicInfoValue(topic_name: str, construction_order: int)

Bases: ValueObject

PublishTopic info.

property construction_order: int

Get construction order.

Returns

int: Construction order.

property topic_name: str

Get topic name.

Returns

str: Topic name.

class caret_analyze.value_objects.PublisherStructValue(node_name: str, topic_name: str, callback_values: tuple[CallbackStructValue, ...] | None, construction_order: int)

Bases: ValueObject, Summarizable

Structured publisher value.

property callback_names: tuple[str, ...] | None

Get callback names.

Returns

tuple[str, …] | None: Callback names.

property callbacks: tuple[CallbackStructValue, ...] | None

Get callbacks.

Returns

tuple[CallbackStructValue, …] | None: Callbacks.

property construction_order: int

Get construction order.

Returns

int: Construction order.

property node_name: str

Get node name.

Returns

str: Node name.

property summary: Summary

Get summary.

Returns

Summary: Summary.

property topic_name: str

Get topic name.

Returns

str: Topic name.

class caret_analyze.value_objects.PublisherValue(topic_name: str, node_name: str, node_id: str, callback_ids: tuple[str, ...] | None, construction_order: int)

Bases: ValueObject

Publisher value.

property callback_ids: tuple[str, ...] | None

Get callback ids.

Returns

tuple[str, …]: Callback ids.

property construction_order: int

Get construction order.

Returns

int: Construction order.

property node_id: str

Get node id.

Returns

str: Node id.

property node_name: str

Get node name.

Returns

str: Node id.

property topic_name: str

Get topic name.

Returns

str: Topic name.

class caret_analyze.value_objects.Qos(depth: int)

Bases: ValueObject, Summarizable

Qos info.

Note:

Currently, only DEPTH is supported, but it is desirable to support other QoS by adding trace points.

property depth: int

Get depth.

Returns

int: Depth.

property summary: Summary

Get summary.

Returns

Summary: Summary.

class caret_analyze.value_objects.ServiceCallbackStructValue(node_name: str, symbol: str, service_name: str, publish_topics: tuple[PublishTopicInfoValue, ...] | None, construction_order: int, callback_name: str)

Bases: CallbackStructValue, ValueObject

Structured service callback value.

property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: Callback type.

property summary: Summary

Get summary.

Returns

Summary: Summary about value objects and runtime data objects.

class caret_analyze.value_objects.ServiceCallbackValue(callback_id: str, node_name: str, node_id: str, symbol: str, service_name: str, publish_topics: tuple[PublishTopicInfoValue, ...] | None, construction_order: int, *, callback_name: str | None = None)

Bases: CallbackValue

Value object class for representing a service.

property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: Callback type.

property service_name: str

Get service name.

Returns

str: Service name.

class caret_analyze.value_objects.ServiceStructValue(node_name: str, service_name: str, callback_info: ServiceCallbackStructValue | None, construction_order: int)

Bases: ValueObject, Summarizable

Service info.

property callback: ServiceCallbackStructValue | None

Get callback.

Returns

ServiceCallbackStructValue | None: Callback.

property callback_name: str | None

Get callback name.

Returns

str | None: Callback name.

property construction_order: int

Get construction order.

Returns

int: Construction order.

property node_name: str

Get node name.

Returns

str: Node name.

property service_name: str

Get node name.

Returns

str: Node name.

property summary: Summary

Get summary.

Returns

Summary: Summary about value objects and runtime data objects.

class caret_analyze.value_objects.ServiceValue(service_name: str, node_name: str, node_id: str | None, callback_id: str | None, construction_order: int)

Bases: ValueObject

Service info.

property callback_id: str | None

Get callback id.

Returns

str | None: Callback unique id.

property construction_order: int

Get construction order.

Returns

int: Construction order.

property node_id: str | None

Get node id.

Returns

str | None: Node id.

property node_name: str

Get node name.

Returns

str: Node name.

property service_name: str

Get node name.

Returns

str: Node name.

class caret_analyze.value_objects.SubscriptionCallbackStructValue(node_name: str, symbol: str, subscribe_topic_name: str, publish_topics: tuple[PublishTopicInfoValue, ...] | None, construction_order: int, callback_name: str)

Bases: CallbackStructValue, ValueObject

Structured subscription callback value.

property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: Callback type.

property subscribe_topic_name: str

Get subscription topic name.

Returns

str: Topic name which the callback subscribes.

property summary: Summary

Get summary.

Returns

Summary: Summary about value objects and runtime data objects.

class caret_analyze.value_objects.SubscriptionCallbackValue(callback_id: str, node_name: str, node_id: str, symbol: str, subscribe_topic_name: str, publish_topics: tuple[PublishTopicInfoValue, ...] | None, construction_order: int, *, callback_name: str | None = None)

Bases: CallbackValue

Subscription callback value.

property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: Callback type.

property subscribe_topic_name: str

Get subscription topic name.

Returns

str: Topic name which the callback subscribes.

class caret_analyze.value_objects.SubscriptionStructValue(node_name: str, topic_name: str, callback_info: SubscriptionCallbackStructValue | None, construction_order: int)

Bases: ValueObject, Summarizable

Subscription info.

property callback: SubscriptionCallbackStructValue | None

Get callback.

Returns

SubscriptionCallbackStructValue | None: Callback.

property callback_name: str | None

Get callback name.

Returns

str | None: Callback name

property construction_order: int

Get construction order.

Returns

int: Construction order.

property node_name: str

Get node name.

Returns

str: Node name.

property summary: Summary

Get summary.

Returns

Summary: Summary info.

property topic_name: str

Get a topic name.

Returns

str: Topic name of this subscription.

class caret_analyze.value_objects.SubscriptionValue(topic_name: str, node_name: str, node_id: str | None, callback_id: str | None, construction_order: int)

Bases: ValueObject

Subscription info.

property callback_id: str | None

Get callback id.

Returns

str | None: Callback unique id.

property construction_order: int

Get construction order.

Returns

int: Construction order.

property node_id: str | None

Get node id.

Returns

str | None: Node id.

property node_name: str

Get node name.

Returns

str: Node name.

property topic_name: str

Get a topic name.

Returns

str: Topic name of this communication.

class caret_analyze.value_objects.Tilde(node_name: str, message_context_dict: dict, subscription: SubscriptionStructValue | None, publisher: PublisherStructValue | None, callbacks: tuple[CallbackStructValue, ...] | None)

Bases: MessageContext

TYPE_NAME = 'tilde'

Tilde.

Latency is calculated from tilde.

property context_type: MessageContextType

Get context type.

Returns

MessageContextType: Message context type.

is_applicable_path(subscription: SubscriptionStructValue | None, publisher: PublisherStructValue | None, callbacks: tuple[CallbackStructValue, ...] | None) → bool

Get applicable path.

Parameters

subscriptionSubscriptionStructValue | None: Target subscription value.
publisherPublisherStructValue | None: Target publisher value.
callbackstuple[CallbackStructValue, …] | None: Target callbacks.

Returns

bool: True if applicable path, false otherwise.

verify() → bool

Get verify.

Returns

bool: Verify or not.

class caret_analyze.value_objects.TimerCallbackStructValue(node_name: str, symbol: str, period_ns: int, publish_topics: tuple[PublishTopicInfoValue, ...] | None, construction_order: int, callback_name: str)

Bases: CallbackStructValue, ValueObject

Structured timer callback value.

property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: Callback type.

property period_ns: int

Get period.

Returns

int: Period of the timer.

property summary: Summary

Get summary.

Returns

Summary: Summary about value objects and runtime data objects.

class caret_analyze.value_objects.TimerCallbackValue(callback_id: str, node_name: str, node_id: str, symbol: str, period_ns: int, publish_topics: tuple[PublishTopicInfoValue, ...] | None, construction_order: int, *, callback_name: str | None = None)

Bases: CallbackValue

Value object class for representing a timer.

This class has minimal information and no structure, and used as the return value of ArchitectureReader.

property callback_type: CallbackType

Get callback type name.

Returns

CallbackType: Callback type.

property period_ns: int

Get period.

Returns

int: Period of the timer.

class caret_analyze.value_objects.TimerStructValue(node_name: str, period_ns: int, callback_info: TimerCallbackStructValue | None, construction_order: int)

Bases: ValueObject, Summarizable

Timer info.

property callback: TimerCallbackStructValue | None

Get callback.

Returns

TimerCallbackStructValue | None: Callback.

property callback_name: str | None

Get callback name.

Returns

str | None: Callback name.

property construction_order: int | None

Get construction order.

Returns

int | None: Construction order.

property node_name: str

Get node name.

Returns

str: Node name.

property period_ns: int

Get period.

Returns

int: Period of the timer.

property summary: Summary

Get summary.

Returns

Summary: Summary info.

class caret_analyze.value_objects.TimerValue(period: int, node_name: str, node_id: str | None, callback_id: str | None, construction_order: int)

Bases: ValueObject

Timer info.

property callback_id: str | None

Get callback id.

Returns

str | None: Callback unique id.

property construction_order: int

Get construction order.

Returns

int: Construction order.

property node_id: str | None

Get node id.

Returns

str | None: Node id.

property node_name: str

Get node name.

Returns

str: Node name.

property period: int

Get period.

Returns

int: Period of the timer.

class caret_analyze.value_objects.UseLatestMessage(node_name: str, message_context_dict: dict, subscription: SubscriptionStructValue | None, publisher: PublisherStructValue | None, child: tuple[CallbackStructValue, ...] | None)

Bases: MessageContext

TYPE_NAME = 'use_latest_message': Use message context

property context_type: MessageContextType

Get context type.

Returns

MessageContextType: Message context type.

verify() → bool

Get verify.

Returns

bool: Verify or not.

class caret_analyze.value_objects.ValueObject

Bases: object

Value object base class.

Classes that inherit from this class will become immutable ValueObjects, and __eq__ and __hash__ will be calculated based on public properties.

Note:

Since the hash value is immutable, inherited classes can be used as a dictionary type key. It is also suitable for cache use and does not unintentionally change properties.

class caret_analyze.value_objects.VariablePassingStructValue(node_name: str, callback_write: CallbackStructValue, callback_read: CallbackStructValue)

Bases: ValueObject, Summarizable

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: CallbackStructValue

Get read-side callback.

Returns

str: Read-side callback.

property callback_write: CallbackStructValue

Get write-side callback.

Returns

str: Write-side callback.

property node_name: str

Get node name.

Returns

str: Node name.

property summary: Summary

Get summary.

Returns

Summary: Summary about value objects and runtime data objects.

class caret_analyze.value_objects.VariablePassingValue(node_name: str, callback_id_write: str, callback_id_read: str)

Bases: ValueObject

variable passing info.

property callback_id_read: str

Get read-side callback id.

Returns

str: Read-side callback id.

property callback_id_write: str

Get write-side callback id.

Returns

str: Write-side callback id.

property node_name: str

Get node name.

Returns

str: Node name.