Skip to content

networking.p2p.messages

messages

Classes:

Name Description
Msg

Classes

Msg 

Msg(sender: str | None = None, content: any = None, timestamp_net: str | None = None, channel: str | None = None, content_type: str = MISC, piggyback: str | None = None, _proto_msg: Message = None)

The constructor should be used either to create a new message filling the fields, or to parse an existing Protobuf message (passing _proto_msg). In the latter case, the other fields are ignored and the Protobuf message is used as-is. The message is simply stored in the internal _proto_msg field and other fields can be accessed through properties.

Methods:

Name Description
to_bytes

Serializes the internal Protobuf message to bytes.
from_bytes

Deserializes a byte array into a new Msg instance.

Attributes:

Name Type Description
content any

The main content of the message, decoded on-the-fly with caching.
Source code in unaiverse/networking/p2p/messages.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(self,
             sender: str | None = None,
             content: any = None,
             timestamp_net: str | None = None,
             channel: str | None = None,
             content_type: str = MISC,
             piggyback: str | None = None,
             _proto_msg: pb.Message = None):
    """The constructor should be used either to create a new message filling the fields,
    or to parse an existing Protobuf message (passing _proto_msg). In the latter case,
    the other fields are ignored and the Protobuf message is used as-is. The message is
    simply stored in the internal `_proto_msg` field and other fields can be accessed
    through properties."""

    self._decoded_content: any = None  # Cache for decompressed content

    if _proto_msg is not None:

        # Check if any other arguments were simultaneously provided
        other_args = [sender, content, timestamp_net, channel, piggyback]
        if any(arg is not None for arg in other_args):
            raise ValueError("Cannot specify other arguments when creating a Msg from a _proto_msg.")

        # This path is used by from_bytes, message is already built
        self._proto_msg = _proto_msg
        return

    # Sanity checks
    assert sender is not None, "Sender must be specified for a new message."
    assert isinstance(sender, str), "Sender must be a string"
    assert timestamp_net is None or isinstance(timestamp_net, str), "Invalid timestamp_net"
    assert channel is None or isinstance(channel, str), "Invalid channel"
    assert content_type in Msg.CONTENT_TYPES, "Invalid content type"

    # --- SMART CONSTRUCTOR: Populates the correct 'oneof' field ---
    self._proto_msg = pb.Message()
    self._proto_msg.sender = sender if sender is not None else ""
    self._proto_msg.timestamp_net = timestamp_net if timestamp_net is not None else \
        datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")
    self._proto_msg.content_type = content_type if content_type is not None else self.MISC
    self._proto_msg.channel = channel if channel is not None else "<unknown>"
    self._proto_msg.piggyback = piggyback if piggyback is not None else ""

    if content is None or content == "<empty>":
        return  # Nothing to set in the 'oneof'

    # Route the content to the correct builder
    if content_type == Msg.STREAM_SAMPLE:
        self._build_stream_sample_content(content)
    else:

        # All other structured types use the generic json_content field
        self._build_json_content(content)
Attributes
content property 
content: any

The main content of the message, decoded on-the-fly with caching.

Methods:
to_bytes 
to_bytes() -> bytes

Serializes the internal Protobuf message to bytes.

Source code in unaiverse/networking/p2p/messages.py 
172
173
174
def to_bytes(self) -> bytes:
    """Serializes the internal Protobuf message to bytes."""
    return self._proto_msg.SerializeToString()
from_bytes classmethod 
from_bytes(msg_bytes: bytes) -> Msg

Deserializes a byte array into a new Msg instance.

Source code in unaiverse/networking/p2p/messages.py 
176
177
178
179
180
181
182
183
@classmethod
def from_bytes(cls, msg_bytes: bytes) -> 'Msg':
    """Deserializes a byte array into a new Msg instance."""
    pb_msg = pb.Message()
    pb_msg.ParseFromString(msg_bytes)

    # Pass the parsed protobuf message to the constructor
    return cls(_proto_msg=pb_msg)