Message Protocol#

message_protocol #

PROTOCOL_VERSION `module-attribute` #

PROTOCOL_VERSION = 0

_sgen_namespace `module-attribute` #

_sgen_namespace = SgenNamespace(
    "AllenNeuralDynamics.AindBehaviorServices.MessageProtocol"
)

TPayload `module-attribute` #

TPayload = TypeVar('TPayload', bound=BaseModel)

RegisteredMessages `module-attribute` #

RegisteredMessages = _sgen_namespace.sgen_typename()(
    create_model(
        "RegisteredMessages",
        __base__=_Message[RegisteredPayload],
    )
)

SgenNamespace #

SgenNamespace(namespace)

Source code in src/aind_behavior_services/schema/__init__.py

def __init__(self, namespace: str):
    self._namespace = namespace

namespace `property` #

namespace

sgen_typename #

sgen_typename(*, typename=None)

Source code in src/aind_behavior_services/schema/__init__.py

def sgen_typename(self, *, typename: str | None = None) -> Callable[[Type[T]], Type[T]]:
    return sgen_typename(typename=typename, namespace=self._namespace)

MessageType #

Bases: StrEnum

Enumeration of possible message types in the protocol.

Examples:

MessageType.REQUEST  # 'request'
MessageType.REPLY    # 'reply'
MessageType.EVENT    # 'event'

REQUEST `class-attribute` `instance-attribute` #

REQUEST = 'request'

REPLY `class-attribute` `instance-attribute` #

REPLY = 'reply'

EVENT `class-attribute` `instance-attribute` #

EVENT = 'event'

_Message `pydantic-model` #

Bases: BaseModel, Generic[TPayload]

A generic message container that can carry any payload type. While not marked as abstract, it is intended to be subclassed for specific message types with defined payloads.

ATTRIBUTE	DESCRIPTION
`cls_type`	The specific message class type identifier (discriminator)
`message_type`	The category of message (request, reply, or event) TYPE: `MessageType`
`protocol_version`	The major version of the message protocol being used TYPE: `Literal[PROTOCOL_VERSION]`
`timestamp`	When the message was created TYPE: `Optional[AwareDatetime]`
`payload`	The actual message content TYPE: `SerializeAsAny[TPayload]`
`process_id`	Identifier of the process (e.g.: executable) that created the message TYPE: `Optional[str]`
`hostname`	Name of the host machine that created the message TYPE: `Optional[str]`
`rig_name`	Name of the experimental rig that created the message TYPE: `Optional[str]`

Fields:

message_type (MessageType)
protocol_version (Literal[PROTOCOL_VERSION])
timestamp (Optional[AwareDatetime])
payload (SerializeAsAny[TPayload])
process_id (Optional[str])
hostname (Optional[str])
rig_name (Optional[str])

message_type `pydantic-field` #

message_type

protocol_version `pydantic-field` #

protocol_version = PROTOCOL_VERSION

timestamp `pydantic-field` #

timestamp

The timestamp of the message

payload `pydantic-field` #

payload

The payload of the message

process_id `pydantic-field` #

process_id

Process that created the message

hostname `pydantic-field` #

hostname

Hostname that created the message

rig_name `pydantic-field` #

rig_name

Rig name that created the message

LogLevel #

Bases: IntEnum

Enumeration of log levels for the logging system.

Follows standard Python logging levels with integer values that allow for easy comparison and filtering.

Examples:

LogLevel.ERROR > LogLevel.WARNING  # True
LogLevel.DEBUG.value               # 10
str(LogLevel.INFO)                 # 'LogLevel.INFO'

CRITICAL `class-attribute` `instance-attribute` #

CRITICAL = 50

ERROR `class-attribute` `instance-attribute` #

ERROR = 40

WARNING `class-attribute` `instance-attribute` #

WARNING = 30

INFO `class-attribute` `instance-attribute` #

INFO = 20

DEBUG `class-attribute` `instance-attribute` #

DEBUG = 10

NOTSET `class-attribute` `instance-attribute` #

NOTSET = 0

LogPayload `pydantic-model` #

Bases: BaseModel

Payload for log messages containing logging information.

This payload carries log data including the message content, severity level, optional context, and application version.

ATTRIBUTE	DESCRIPTION
`message`	The actual log message text TYPE: `str`
`level`	Severity level of the log entry TYPE: `LogLevel`
`context`	Optional additional data related to the log TYPE: `Optional[SerializeAsAny[Any]]`
`application_version`	Version of the application generating the log TYPE: `Optional[str]`

Examples:

log_payload = LogPayload(
    message="System startup complete",
    level=LogLevel.INFO,
    context={"operator": "John Doe"},
    application_version="1.0.0"
)
print(log_payload.level)  # LogLevel.INFO

Fields:

payload_type (Literal['LogPayload'])
message (str)
level (LogLevel)
context (Optional[SerializeAsAny[Any]])
application_version (Optional[str])

payload_type `pydantic-field` #

payload_type = 'LogPayload'

message `pydantic-field` #

message

The message of the log

level `pydantic-field` #

level = LogLevel.DEBUG

The level of the log message

context `pydantic-field` #

context = None

Additional context for the log message

application_version `pydantic-field` #

application_version = None

The version of the application

HeartbeatStatus #

Bases: IntEnum

Enumeration of possible heartbeat status values.

Represents the health status of a system component, with higher values indicating more severe issues.

Examples:

HeartbeatStatus.OK                            # <HeartbeatStatus.OK: 0>
HeartbeatStatus.CRITICAL > HeartbeatStatus.WARNING  # True
int(HeartbeatStatus.ERROR)                    # 2

OK `class-attribute` `instance-attribute` #

OK = 0

WARNING `class-attribute` `instance-attribute` #

WARNING = 1

ERROR `class-attribute` `instance-attribute` #

ERROR = 2

CRITICAL `class-attribute` `instance-attribute` #

CRITICAL = 3

HeartbeatPayload `pydantic-model` #

Bases: BaseModel

Payload for heartbeat messages indicating system health status.

Heartbeat messages are used to monitor the health and availability of system components. They include a status indicator and optional context information.

ATTRIBUTE	DESCRIPTION
`context`	Optional additional data about the system state TYPE: `SerializeAsAny[Optional[Any]]`
`status`	Current health status of the component TYPE: `HeartbeatStatus`

Examples:

heartbeat = HeartbeatPayload(
    status=HeartbeatStatus.OK,
    context={"cpu_usage": 0.25, "memory_usage": 0.60}
)
print(heartbeat.status)  # HeartbeatStatus.OK

warning_heartbeat = HeartbeatPayload(
    status=HeartbeatStatus.WARNING,
    context={"disk_space_low": True}
)

Fields:

payload_type (Literal['HeartbeatPayload'])
context (SerializeAsAny[Optional[Any]])
status (HeartbeatStatus)

payload_type `pydantic-field` #

payload_type = 'HeartbeatPayload'

context `pydantic-field` #

context = None

Additional context for the heartbeat message.

status `pydantic-field` #

status

The status of the heartbeat message

RegisteredPayload #

Bases: RootModel

root `instance-attribute` #

root

model_config `class-attribute` `instance-attribute` #

model_config = ConfigDict(
    json_schema_extra={"x-abstract": True}
)

Message #

Bases: RootModel

root `instance-attribute` #

root