networking.p2p

p2p

Modules:

Name	Description
golibp2p
lib_types
messages
mylogger
p2p

Classes:

Name	Description
P2P	Python wrapper for the Go libp2p shared library.
P2PError	Custom exception class for P2P library errors.
TypeInterface	Helper class for converting between Python types and Go types using ctypes.

Classes

P2P

P2P(port: int = 0, ips: List[str] = None, enable_relay_client: bool = True, enable_relay_service: bool = False, wait_public_reachability: bool = False, max_connections: int = 1000)

Python wrapper for the Go libp2p shared library.

This class initializes a libp2p node, provides methods to interact with the p2p network (connect, send/receive messages, pubsub, relay), and manages the lifecycle of the underlying Go node.

Attributes:

Name	Type	Description
libp2p	LibP2P	Static class attribute holding the loaded Go library instance. Must be set before instantiating P2P. Example: P2P.libp2p = LibP2P()
peer_id	str	The Peer ID of the initialized local node.
addresses	Optional[List[str]]	List of multiaddresses the local node is listening on.
is_public	bool	Whether the node is publicly reachable.
peer_map	Dict[str, Any]	A dictionary to potentially store information about connected peers (managed manually or by polling thread).

Initializes and starts a new libp2p node.

Parameters:

Name	Type	Description	Default
port	int	The (first) TCP port to listen on (0 for random).	0
ips	List[str]	A list of specific IP addresses to listen on. Defaults to ["0.0.0.0"].	None
enable_relay_client	bool	Enable listening to relayed connections for this node.	True
enable_relay_service	bool	Enable relay service capabilities for this node.	False
wait_public_reachability	bool	Tries every possible attempt to make the node publicly reachable (UPnP, HolePunching, AutoNat via DHT...).	False
max_connections	int	Maximum number of connections this node can handle.	1000

Raises:

Type	Description
P2PError	If the node creation fails in the Go library.
AttributeError	If P2P.libp2p has not been set before instantiation.

Methods:

Name	Description
setup_library	Initializes the underlying Go library. Must be called once. This is called automatically.
connect_to	Establishes a connection with a remote peer.
disconnect_from	Closes connections to a specific peer and removes tracking.
send_message_to_peer	Sends a direct message to a specific peer.
broadcast_message	Broadcasts a message using PubSub to the node's own topic.
pop_messages	Retrieves and removes the first message from the queue of each channel for this node instance.
subscribe_to_topic	Subscribes to a PubSub topic to receive messages.
unsubscribe_from_topic	Unsubscribes from a PubSub topic.
reserve_on_relay	Attempts to reserve a slot on a specified relay node.
get_node_addresses	Gets the known multiaddresses for the local node or a specific peer.
get_connected_peers_info	Gets information about currently connected peers from the Go library.
get_rendezvous_peers_info	Gets the full rendezvous state from the Go library, including peers and metadata.
get_message_queue_length	Gets the current number of messages in the incoming queue.
close	Gracefully shuts down the libp2p node and stops background threads.

Source code in unaiverse/networking/p2p/p2p.py

def __init__(self,
             port: int = 0,
             ips: List[str] = None,
             enable_relay_client: bool = True,
             enable_relay_service: bool = False,
             wait_public_reachability: bool = False,
             max_connections: int = 1000,
             ) -> None:
    """
    Initializes and starts a new libp2p node.

    Args:
        port: The (first) TCP port to listen on (0 for random).
        ips: A list of specific IP addresses to listen on. Defaults to ["0.0.0.0"].
        enable_relay_client: Enable listening to relayed connections for this node.
        enable_relay_service: Enable relay service capabilities for this node.
        wait_public_reachability: Tries every possible attempt to make the node publicly reachable (UPnP, HolePunching, AutoNat via DHT...).
        max_connections: Maximum number of connections this node can handle.

    Raises:
        P2PError: If the node creation fails in the Go library.
        AttributeError: If P2P.libp2p has not been set before instantiation.
    """

    # --- CRITICAL: Check if library is initialized ---
    if not P2P._library_initialized:
        raise P2PError("P2P library not set up. Call P2P.setup_library() before creating an instance.")

    # Assign instance ID
    assigned_instance_id = -1
    with P2P._instance_lock:
        for _instance_id, i in enumerate(self._instance_ids):
            if not i:
                self._instance_ids[_instance_id] = True
                assigned_instance_id = _instance_id
                break
        if assigned_instance_id == -1:
            raise P2PError(
                f"Cannot create new P2P instance: Maximum number of instances "
                f"({P2P._MAX_INSTANCES})."
            )

    self._instance: int = assigned_instance_id
    logger.info(f"🚀 Attempting to initialize P2P Node with auto-assigned Instance ID: {self._instance}")

    self._port = port
    self._ips = ips if ips is not None else []
    self._enable_relay_client = enable_relay_client or enable_relay_service
    self._enable_relay_service = enable_relay_service
    self._wait_public_reachability = wait_public_reachability
    self._max_connections = max_connections
    self._peer_id: Optional[str] = None
    self._peer_map: Dict[str, Any] = {}  # Map to store peer info {peer_id: info}
    self._poll_interval = 5.0  # Polling interval for background threads
    self._stop_event = threading.Event()

    logger.info(f"🐍 Creating Node (Instance ID: {self._instance})...")
    try:

        # Call the Go function
        result_ptr = P2P.libp2p.CreateNode(
            P2P._type_interface.to_go_int(self._instance),
            P2P._type_interface.to_go_int(port),
            P2P._type_interface.to_go_json(self._ips),
            P2P._type_interface.to_go_bool(enable_relay_client),
            P2P._type_interface.to_go_bool(enable_relay_service),
            P2P._type_interface.to_go_bool(wait_public_reachability),
            P2P._type_interface.to_go_int(max_connections),
        )
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            err_msg = "Received null result from Go CreateNode."
            logger.error(f"[Instance {self._instance}] {err_msg}")
            raise P2PError(f"[Instance {self._instance}] {err_msg}")
        if result.get('state') == "Error":
            err_msg = result.get('message', 'Unknown Go error on CreateNode')
            logger.error(f"[Instance {self._instance}] Go error: {err_msg}")
            raise P2PError(f"[Instance {self._instance}] Failed to create node: {err_msg}")

        message_data = result.get('message')
        initial_addresses = message_data.get("addresses", [])
        self._is_public = message_data.get("isPublic", False)

        # Check teh returned data
        if not isinstance(initial_addresses, list) or not initial_addresses:
            err_msg = "Received empty or invalid addresses list from Go CreateNode."
            logger.error(f"[Instance {self._instance}] {err_msg}")
            raise P2PError(f"[Instance {self._instance}] {err_msg}")

        self._peer_id = initial_addresses[0].split("/")[-1]

        # Cache for the dynamic addresses property
        self._address_cache: Optional[List[str]] = initial_addresses
        self._address_cache_time: float = time.monotonic()

        # Re-ordering/sorting, discarding /p2p-circuit/ because right now it is useless and preferring /udp/
        addresses_quic = [a for a in self._address_cache if "/quic-v1/" in a]
        addresses_webrtc = [a for a in self._address_cache if "/webrtc" in a]
        addresses_tcp = [a for a in self._address_cache if "/tcp/" in a]
        addresses = addresses_quic + addresses_webrtc + addresses_tcp
        self._address_cache.clear()
        for _addr in addresses:
            self._address_cache.append(_addr)

        logger.info(f"✅ [Instance {self._instance}] Node created with ID: {self._peer_id}")
        logger.info(f"👂 [Instance {self._instance}] Listening on: {self._address_cache}")
        logger.info(f"🌐 [Instance {self._instance}] Publicly reachable: {self._is_public}")

        # Start background threads if desired (currently commented out in original)
        # logger.info(f"[Instance {self._instance}] 🧵 Starting background polling threads...")
        # self._get_connected_peers_thread = threading.Thread(...)
        # self._get_connected_peers_thread.start()
        # self._check_message_queue_thread = threading.Thread(...)
        # self._check_message_queue_thread.start()

        logger.info(f"🎉 [Instance {self._instance}] Node initialized successfully.")

    except Exception as e:
        logger.error(f"❌ [Instance {self._instance}] Node creation failed: {e}")

        # Reclaim the instance ID using the _instance_ids list
        if self._instance != -1:  # Check if an ID was actually assigned
            with P2P._instance_lock:
                P2P._instance_ids[self._instance] = False
                logger.info(f"[Instance {self._instance}] Reclaimed instance ID {self._instance} due to creation failure.")
        raise  # Re-raise the exception that caused the failure

    # # --- Background Threads ---
    self._stop_event = threading.Event()  # Event to signal threads to stop

    logger.info("🎉 Node created successfully and background polling started.")

Attributes

peer_id property

peer_id: Optional[str]

Returns the Peer ID of the local node.

addresses property

addresses: Optional[List[str]]

Returns the list of multiaddresses the local node is listening on. This property queries the Go layer directly, caching the result for performance.

is_public property

is_public: Optional[bool]

Returns a boolean stating whether the local node is publicly reachable.

relay_is_enabled property

relay_is_enabled: bool

Returns whether the relay client functionality is enabled for this node.

Methods:

setup_library classmethod

setup_library(max_instances: Optional[int] = None, max_channels: Optional[int] = None, max_queue_per_channel: Optional[int] = None, max_message_size: Optional[int] = None, enable_logging: bool = False) -> None

Initializes the underlying Go library. Must be called once. This is called automatically.

Source code in unaiverse/networking/p2p/p2p.py

@classmethod
def setup_library(cls,
                  max_instances: Optional[int] = None,
                  max_channels: Optional[int] = None,
                  max_queue_per_channel: Optional[int] = None,
                  max_message_size: Optional[int] = None,
                  enable_logging: bool = False) -> None:
    """
    Initializes the underlying Go library. Must be called once. This is called automatically.
    """
    with cls._initialize_lock:
        if cls._library_initialized:
            logger.warning("P2P library is already initialized. Skipping setup.")
            return

        if not hasattr(cls, 'libp2p') or cls.libp2p is None:
            raise P2PError("Library not loaded before setup. Check package __init__.py")

        # Configure Python logging based on the flag
        if not enable_logging:
            logger.setLevel(logging.WARNING)
        else:
            logger.setLevel(logging.INFO)

        logger.info("🐍 Setting up and initializing P2P library core with user settings...")
        cls._type_interface = TypeInterface(cls.libp2p)

        # Use provided arguments or fall back to class defaults
        _max_instances = max_instances if max_instances is not None else cls._MAX_INSTANCES
        _max_channels = max_channels if max_channels is not None else cls._MAX_NUM_CAHNNELS
        _max_queue = max_queue_per_channel if max_queue_per_channel is not None else cls._MAX_QUEUE_PER_CHANNEL
        _max_msg_size = max_message_size if max_message_size is not None else cls._MAX_MESSAGE_SIZE

        # Update class attributes if they were overridden
        cls._MAX_INSTANCES = _max_instances
        cls._instance_ids = [False, ] * _max_instances  # Resize the tracking list

        # Call the Go function to set up its internal state
        logger.info("🐍 Initializing Go library core...")
        cls.libp2p.InitializeLibrary(
            cls._type_interface.to_go_int(_max_instances),
            cls._type_interface.to_go_int(_max_channels),
            cls._type_interface.to_go_int(_max_queue),
            cls._type_interface.to_go_int(_max_msg_size),
            cls._type_interface.to_go_bool(enable_logging)
        )

        cls._library_initialized = True
        logger.info("✅ Go library initialized successfully.")

connect_to

connect_to(multiaddrs: list[str]) -> Dict[str, Any]

Establishes a connection with a remote peer.

Parameters:

Name	Type	Description	Default
multiaddrs	list[str]	The list of multiaddress strings of the peer to try to connect to.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary containing the connected peer's AddrInfo (ID and Addrs).

Raises:

Type	Description
P2PError	If the connection fails.
ValueError	If the multiaddr is invalid.

Source code in unaiverse/networking/p2p/p2p.py

def connect_to(self, multiaddrs: list[str]) -> Dict[str, Any]:
    """
    Establishes a connection with a remote peer.

    Args:
        multiaddrs: The list of multiaddress strings of the peer to try to connect to.

    Returns:
        A dictionary containing the connected peer's AddrInfo (ID and Addrs).

    Raises:
        P2PError: If the connection fails.
        ValueError: If the multiaddr is invalid.
    """
    if not multiaddrs or not isinstance(multiaddrs, list):
        logger.error("Invalid multiaddr provided.")
        raise ValueError("Invalid multiaddr provided.")
    dest_peer_id = multiaddrs[0].split('/')[-1]
    logger.info(f"📞 Attempting to connect to: {dest_peer_id}...")
    try:
        result_ptr = P2P.libp2p.ConnectTo(
            P2P._type_interface.to_go_int(self._instance),
            P2P._type_interface.to_go_json(multiaddrs)
            )
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error("Failed to connect to peer, received null result.")
            raise P2PError("Failed to connect to peer, received null result.")
        if result.get('state') == "Error":
            logger.error(f"Failed to connect to peer '{dest_peer_id}': {result.get('message', 'Unknown Go error')}")
            raise P2PError(f"Failed to connect to peer '{dest_peer_id}': {result.get('message', 'Unknown Go error')}")

        peer_info = result.get('message', {})
        logger.info(f"✅ Connection initiated to peer: {peer_info.get('ID', dest_peer_id)}")  # Use ID if available

        # Optionally update internal peer map here
        # self._peer_map[peer_info.get('ID')] = peer_info
        return peer_info

    except Exception as e:
        logger.error(f"❌ Connection to {dest_peer_id} failed: {e}")
        raise P2PError(f"Connection to {dest_peer_id} failed") from e

disconnect_from

disconnect_from(peer_id: str) -> None

Closes connections to a specific peer and removes tracking.

Parameters:

Name	Type	Description	Default
peer_id	str	The Peer ID string of the peer to disconnect from.	required

Raises:

Type	Description
P2PError	If disconnecting fails.
ValueError	If the peer_id is invalid.

Source code in unaiverse/networking/p2p/p2p.py

def disconnect_from(self, peer_id: str) -> None:
    """
    Closes connections to a specific peer and removes tracking.

    Args:
        peer_id: The Peer ID string of the peer to disconnect from.

    Raises:
        P2PError: If disconnecting fails.
        ValueError: If the peer_id is invalid.
    """
    if not peer_id or not isinstance(peer_id, str):
        logger.error("Invalid Peer ID provided.")
        raise ValueError("Invalid Peer ID provided.")

    # Basic peer ID format check (Qm... or 12D3...)
    if not (peer_id.startswith("Qm") or peer_id.startswith("12D3")):
        logger.warning(f"⚠️ Warning: Peer ID '{peer_id}' does not look like a standard v0 or v1 ID.")

    logger.info(f"🔌 Attempting to disconnect from peer: {peer_id}...")
    try:
        result_ptr = P2P.libp2p.DisconnectFrom(
            P2P._type_interface.to_go_int(self._instance),
            P2P._type_interface.to_go_string(peer_id)
            )
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error("Failed to disconnect from peer, received null result.")
            raise P2PError("Failed to disconnect from peer, received null result.")

        if result.get('state') == "Error":
            logger.error(f"Failed to disconnect from peer '{peer_id}': {result.get('message', 'Unknown Go error')}")
            raise P2PError(f"Failed to disconnect from peer '{peer_id}': {result.get('message', 'Unknown Go error')}")

        logger.info(f"✅ Successfully disconnected from {peer_id}")

        # Optionally remove from internal peer map
        # if peer_id in self._peer_map: del self._peer_map[peer_id]

    except Exception as e:
        logger.error(f"❌ Disconnection from {peer_id} failed: {e}")
        raise P2PError(f"Disconnection from {peer_id} failed") from e

send_message_to_peer

send_message_to_peer(channel: str, msg: Msg) -> None

Sends a direct message to a specific peer.

Parameters:

Name	Type	Description	Default
channel	str	The string identifying the channel for the communication.	required
msg	Msg	The message to send (Msg).	required

Raises:

Type	Description
P2PError	If message sending fails (based on return code).
ValueError	If inputs are invalid.
TypeError	If data is not bytes.

Source code in unaiverse/networking/p2p/p2p.py

def send_message_to_peer(self, channel: str, msg: Msg) -> None:
    """
    Sends a direct message to a specific peer.

    Args:
        channel: The string identifying the channel for the communication.
        msg: The message to send (Msg).

    Raises:
        P2PError: If message sending fails (based on return code).
        ValueError: If inputs are invalid.
        TypeError: If data is not bytes.
    """
    if not channel or not isinstance(channel, str):
        logger.error("Invalid channel provided.")
        raise ValueError("Invalid channel provided.")

    if not isinstance(msg, Msg):
        logger.error("Invalid message provided (must be of type Msg).")
        raise ValueError("Invalid message provided (must be of type Msg).")

    # Serialize the entire message object to bytes using Protobuf.
    payload_bytes = msg.to_bytes()
    payload_len = len(payload_bytes)
    msg_content_type = msg.content_type
    peer_id = channel.split("::dm:")[1].split('-')[0]  # Extract Peer ID from channel format

    logger.info(f"📤 Sending message (type: {msg_content_type}, len: {payload_len}) to peer: {peer_id}...")

    # Call the Go function
    try:
        result_ptr = P2P.libp2p.SendMessageToPeer(
            P2P._type_interface.to_go_int(self._instance),
            P2P._type_interface.to_go_string(channel),
            P2P._type_interface.to_go_bytes(payload_bytes),  # Pass bytes directly
            P2P._type_interface.to_go_int(payload_len),
        )
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error(f"Failed to send direct message to {peer_id}, received null result.")
            raise P2PError(f"Failed to send direct message to {peer_id}, received null result.")

        if result.get('state') == "Error":
            logger.error(f"Failed to send direct message to '{peer_id}': {result.get('message', 'Unknown Go error')}")
            raise P2PError(f"Failed to send direct message to '{peer_id}': {result.get('message', 'Unknown Go error')}")

        logger.info(f"✅ Message sent successfully to {peer_id}.")

    except Exception as e:
        logger.error(f"❌ Sending direct message to {peer_id} failed: {e}")
        raise P2PError(f"Sending direct message to {peer_id} failed") from e

broadcast_message

broadcast_message(channel: str, msg: Msg) -> None

Broadcasts a message using PubSub to the node's own topic. Peers subscribed to this node's Peer ID topic will receive it.

Parameters:

Name	Type	Description	Default
channel	str	The Channel for this topic (e.g., owner_peer_id:topic_name).	required
msg	Msg	The message to send (Msg).	required

Raises:

Type	Description
P2PError	If broadcasting fails.
ValueError	If inputs are invalid.
TypeError	If data is not bytes.

Source code in unaiverse/networking/p2p/p2p.py

def broadcast_message(self, channel: str, msg: Msg) -> None:
    """
    Broadcasts a message using PubSub to the node's own topic.
    Peers subscribed to this node's Peer ID topic will receive it.

    Args:
        channel: The Channel for this topic (e.g., owner_peer_id::ps:topic_name).
        msg: The message to send (Msg).

    Raises:
        P2PError: If broadcasting fails.
        ValueError: If inputs are invalid.
        TypeError: If data is not bytes.
    """
    if not channel or not isinstance(channel, str):
        raise ValueError("Invalid channel provided.")
    if not isinstance(msg, Msg):
        raise ValueError("Invalid message provided (must be of type Msg).")

    # Serialize the entire message object to bytes using Protobuf.
    payload_bytes = msg.to_bytes()
    payload_len = len(payload_bytes)
    msg_content_type = msg.content_type

    logger.info(f"📢 Broadcasting message (type: {msg_content_type}, len: {payload_len}) to own topic ({self.peer_id})...")

    # Call SendMessageToPeer with an empty peer_id string for broadcast
    try:
        result_ptr = P2P.libp2p.SendMessageToPeer(
            P2P._type_interface.to_go_int(self._instance),
            P2P._type_interface.to_go_string(channel),
            P2P._type_interface.to_go_bytes(payload_bytes),
            P2P._type_interface.to_go_int(payload_len),
        )

        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error(f"Failed to broadcast message on channel {channel}, received null result.")
            raise P2PError(f"Failed to broadcast message on channel {channel}, received null result.")

        if result.get('state') == "Error":
            logger.error(f"Failed to broadcast message on channel '{channel}': {result.get('message', 'Unknown Go error')}")
            raise P2PError(f"Failed to broadcast message on channel '{channel}': {result.get('message', 'Unknown Go error')}")

    except Exception as e:
        logger.error(f"❌ Broadcasting to {channel} failed: {e}")
        raise P2PError(f"Broadcasting to {channel} failed") from e

    logger.info(f"✅ Message broadcast successfully to {channel}.")

pop_messages

pop_messages() -> List[Msg]

Retrieves and removes the first message from the queue of each channel for this node instance.

Returns:

Type	Description
List[Msg]	A list of Msg objects. Returns an empty list if no messages were available.

Raises:

Type	Description
P2PError	If popping messages failed internally in Go, or if data conversion fails for any message.

Source code in unaiverse/networking/p2p/p2p.py

def pop_messages(self) -> List[Msg]:
    """
    Retrieves and removes the first message from the queue of each channel for this node instance.

    Returns:
        A list of Msg objects. Returns an empty list if no messages were available.

    Raises:
        P2PError: If popping messages failed internally in Go, or if data
                  conversion fails for any message.
    """
    logger.debug(f"[Instance {self._instance}] Popping message(s)...")
    try:
        go_instance_c = P2P._type_interface.to_go_int(self._instance)

        result_ptr = P2P.libp2p.PopMessages(go_instance_c)

        # From_go_ptr_to_json should handle freeing result_ptr
        raw_result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if raw_result is None:

            # This indicates an issue with the C call or JSON conversion in TypeInterface
            logger.error(f"[Instance {self._instance}] PopMessages: Received null/invalid result from TypeInterface.")
            raise P2PError(f"[Instance {self._instance}] PopMessages: Failed to get valid JSON response.")

        # Check for Go-side error or empty states first
        if isinstance(raw_result, dict):
            state = raw_result.get('state')
            if state == "Empty":
                logger.debug(f"[Instance {self._instance}] PopMessages: Queue is empty.")
                return []  # No messages available
            if state == "Error":
                error_message = raw_result.get('message', 'Unknown Go error during PopMessages')
                logger.error(f"[Instance {self._instance}] PopMessages: {error_message}")
                raise P2PError(f"[Instance {self._instance}] PopMessages: {error_message}")

            # If it's a dict but not a known state, it's unexpected
            logger.warning(f"[Instance {self._instance}] PopMessages: Unexpected dictionary format: {raw_result}")
            raise P2PError(f"[Instance {self._instance}] PopMessages: Unexpected dictionary response format.")

        # Expecting a list of messages if not an error/empty dict
        if not isinstance(raw_result, list):

            # This also covers the case where n=0 and Go returns "[]" which json.loads makes a list
            # If it's not a list at this point, it's an unexpected format.
            logger.error(f"[Instance {self._instance}] PopMessages: Unexpected response format, expected a list or "
                         f"specific state dictionary. Got: {type(raw_result)}")
            raise P2PError(f"[Instance {self._instance}] PopMessages: Unexpected response format.")

        # Process the list of message dictionaries
        processed_messages: List[Msg] = []
        for i, msg_dict in enumerate(raw_result):
            try:
                if not isinstance(msg_dict, dict):
                    logger.warning(f"[Instance {self._instance}] PopMessages: Item {i} in list is not a dict: {msg_dict}")
                    continue  # Skip this malformed entry

                # Extract and validate required fields from the Go message structure
                # Go structure: {"from":"Qm...", "data":"BASE64_ENCODED_DATA"}
                verified_sender_id = msg_dict.get("from")
                base64_data = msg_dict.get("data")

                if not all([verified_sender_id is not None, base64_data is not None]):  # Type can be empty string
                    logger.warning(f"[Instance {self._instance}] PopMessages: Message item {i} missing required fields (from, type, data): {msg_dict}")
                    continue

                # Decode data
                decoded_data = base64.b64decode(base64_data)

                # Attempt to create the higher-level Msg object
                # This assumes Msg.from_bytes can parse your message protocol from decoded_data
                # and that Msg objects store sender, type, channel intrinsically or can be set.
                msg_obj = Msg.from_bytes(decoded_data)

                # --- CRITICAL SECURITY CHECK ---
                # Verify that the sender claimed inside the message payload
                # matches the cryptographically verified sender from the network layer.
                if msg_obj.sender != verified_sender_id:
                    logger.error(f"SENDER MISMATCH! Network sender '{verified_sender_id}' does not match "
                                 f"payload sender '{msg_obj.sender}'. Discarding message.")

                    # In a real-world scenario, you might also want to penalize or disconnect
                    # from a peer that sends such malformed/spoofed messages.
                    continue  # Discard this message

                logger.debug(f"[Instance {self._instance}] Message popped from {verified_sender_id} on {msg_obj.channel}, "
                             f"content_type: {msg_obj.content_type}, len: {len(decoded_data)})")
                processed_messages.append(msg_obj)

            except ValueError as ve:
                logger.error(f"Invalid message created, stopping. Error: {ve}")
                continue  # Skip problematic message
            except (TypeError, binascii.Error) as decode_err:
                logger.error(f"[Instance {self._instance}] PopMessages: Failed to decode Base64 data for a message in batch: {decode_err}. Message dict: {msg_dict}")
                continue  # Skip problematic message
            except Exception as msg_proc_err:  # Catch errors from Msg.from_bytes or attribute setting
                logger.error(f"[Instance {self._instance}] PopMessages: Error processing popped message item {i}: {msg_proc_err}. Message dict: {msg_dict}")
                continue  # Skip problematic message

        if len(raw_result) > 0 and len(processed_messages) == 0:
            logger.warning(f"[Instance {self._instance}] PopMessages: Received {len(raw_result)} messages from Go, but none could be processed into Msg objects.")

            # This could indicate a persistent issue with message format or Msg class.

        return processed_messages

    except P2PError:  # Re-raise P2PError directly
        raise
    except Exception as e:

        # Catch potential JSON parsing errors from TypeInterface or other unexpected errors
        logger.error(f"[Instance {self._instance}] ❌ Error during pop_message: {e}")
        raise P2PError(f"[Instance {self._instance}] Unexpected error during pop_message: {e}") from e

subscribe_to_topic

subscribe_to_topic(channel: str) -> None

Subscribes to a PubSub topic to receive messages.

Parameters:

Name	Type	Description	Default
channel	str	The Channel for this topic (e.g., owner_peer_id:topic_name).	required

Raises:

Type	Description
P2PError	If subscribing fails.
ValueError	If topic_name is invalid.

Source code in unaiverse/networking/p2p/p2p.py

def subscribe_to_topic(self, channel: str) -> None:
    """
    Subscribes to a PubSub topic to receive messages.

    Args:
        channel: The Channel for this topic (e.g., owner_peer_id::ps:topic_name).

    Raises:
        P2PError: If subscribing fails.
        ValueError: If topic_name is invalid.
    """
    if not channel or not isinstance(channel, str):
        logger.error("Invalid topic name provided.")
        raise ValueError("Invalid topic name provided.")
    logger.info(f"<sub> Subscribing to topic: {channel}...")
    try:
        result_ptr = P2P.libp2p.SubscribeToTopic(
            P2P._type_interface.to_go_int(self._instance),
            P2P._type_interface.to_go_string(channel)
            )
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error("Failed to subscribe to topic, received null result.")
            raise P2PError("Failed to subscribe to topic, received null result.")
        if result.get('state') == "Error":
            logger.error(f"Failed to subscribe to topic '{channel}': {result.get('message', 'Unknown Go error')}")
            raise P2PError(f"Failed to subscribe to topic '{channel}': {result.get('message', 'Unknown Go error')}")

        logger.info(f"✅ Successfully subscribed to {channel}")

    except Exception as e:
        logger.error(f"❌ Subscription to {channel} failed: {e}")
        raise P2PError(f"Subscription to {channel} failed") from e

unsubscribe_from_topic

unsubscribe_from_topic(channel: str) -> None

Unsubscribes from a PubSub topic.

Parameters:

Name	Type	Description	Default
channel	str	The Channel for this topic (e.g., owner_peer_id:topic_name).	required

Raises:

Type	Description
P2PError	If unsubscribing fails.
ValueError	If topic_name is invalid.

Source code in unaiverse/networking/p2p/p2p.py

def unsubscribe_from_topic(self, channel: str) -> None:
    """
    Unsubscribes from a PubSub topic.

    Args:
        channel: The Channel for this topic (e.g., owner_peer_id::ps:topic_name).

    Raises:
        P2PError: If unsubscribing fails.
        ValueError: If topic_name is invalid.
    """
    if not channel or not isinstance(channel, str):
        logger.error("Invalid topic name provided.")
        raise ValueError("Invalid topic name provided.")
    logger.info(f"</sub> Unsubscribing from topic: {channel}...")
    try:
        result_ptr = P2P.libp2p.UnsubscribeFromTopic(
            P2P._type_interface.to_go_int(self._instance),
            P2P._type_interface.to_go_string(channel)
            )
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error("Failed to unsubscribe from topic, received null result.")
            raise P2PError("Failed to unsubscribe from topic, received null result.")
        if result.get('state') == "Error":
            logger.error(f"Failed to unsubscribe from topic '{channel}': {result.get('message', 'Unknown Go error')}")
            raise P2PError(f"Failed to unsubscribe from topic '{channel}': {result.get('message', 'Unknown Go error')}")

        logger.info(f"✅ Successfully unsubscribed from {channel}")

    except Exception as e:
        logger.error(f"❌ Unsubscription from {channel} failed: {e}")
        raise P2PError(f"Unsubscription from {channel} failed") from e

reserve_on_relay

reserve_on_relay(relay_peer_id: str) -> str

Attempts to reserve a slot on a specified relay node.

Parameters:

Name	Type	Description	Default
relay_peer_id	str	The peerID of the relay node	required

Returns:

Type	Description
str	The UTC expiration timestamp of the reservation as an ISO 8601 string.

Raises:

Type	Description
P2PError	If the reservation fails.
ValueError	If the relay_multiaddr is invalid.

Source code in unaiverse/networking/p2p/p2p.py

def reserve_on_relay(self, relay_peer_id: str) -> str:
    """
    Attempts to reserve a slot on a specified relay node.

    Args:
        relay_peer_id: The peerID of the relay node

    Returns:
        The UTC expiration timestamp of the reservation as an ISO 8601 string.

    Raises:
        P2PError: If the reservation fails.
        ValueError: If the relay_multiaddr is invalid.
    """
    if not relay_peer_id or not isinstance(relay_peer_id, str):
        logger.error("Invalid relay multiaddr provided.")
        raise ValueError("Invalid relay multiaddr provided.")
    logger.info(f"🅿️ Attempting to reserve slot on relay: {relay_peer_id}...")
    try:
        result_ptr = P2P.libp2p.ReserveOnRelay(
            P2P._type_interface.to_go_int(self._instance),
            P2P._type_interface.to_go_string(relay_peer_id)
            )
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error("Failed to reserve on relay, received null result.")
            raise P2PError("Failed to reserve on relay, received null result.")
        if result.get('state') == "Error":
            logger.error(f"Failed to reserve on relay '{relay_peer_id}': {result.get('message', 'Unknown Go error')}")
            raise P2PError(f"Failed to reserve on relay '{relay_peer_id}': {result.get('message', 'Unknown Go error')}")

        expiration_utc = result.get('message', {})
        if not isinstance(expiration_utc, str):
            raise P2PError(f"Expected expiration timestamp string, but got {type(expiration_utc)}")

        logger.info(f"✅ Reservation successful. Expires at: {expiration_utc}")
        return expiration_utc

    except Exception as e:
        logger.error(f"❌ Reservation on {relay_peer_id} failed: {e}")
        raise P2PError(f"Reservation on {relay_peer_id} failed") from e

get_node_addresses

get_node_addresses(peer_id: str = '') -> List[str]

Gets the known multiaddresses for the local node or a specific peer.

Parameters:

Name	Type	Description	Default
peer_id	str	The Peer ID string of the target peer. If empty, gets addresses for the local node.	''

Returns:

Type	Description
List[str]	A list of multiaddress strings (including the /p2p/PeerID suffix).

Raises:

Type	Description
P2PError	If fetching addresses fails.

Source code in unaiverse/networking/p2p/p2p.py

def get_node_addresses(self, peer_id: str = "") -> List[str]:
    """
    Gets the known multiaddresses for the local node or a specific peer.

    Args:
        peer_id: The Peer ID string of the target peer. If empty, gets
                 addresses for the local node.

    Returns:
        A list of multiaddress strings (including the /p2p/PeerID suffix).

    Raises:
        P2PError: If fetching addresses fails.
    """
    target = "local node" if not peer_id else f"peer {peer_id}"
    logger.info(f"ℹ️ Fetching addresses for {target}...")
    try:
        result_ptr = P2P.libp2p.GetNodeAddresses(
            P2P._type_interface.to_go_int(self._instance),
            P2P._type_interface.to_go_string(peer_id)
            )
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error("Failed to get node addresses, received null result.")
            raise P2PError("Failed to get node addresses, received null result.")
        if result.get('state') == "Error":
            logger.error(f"Failed to get addresses for '{target}': {result.get('message', 'Unknown Go error')}")
            raise P2PError(f"Failed to get addresses for '{target}': {result.get('message', 'Unknown Go error')}")

        addr_list = result.get('message', [])
        logger.info(f"✅ Found addresses for {target}: {addr_list}")
        return addr_list

    except Exception as e:
        logger.error(f"❌ Failed to get addresses for {target}: {e}")
        raise P2PError(f"Failed to get addresses for {target}") from e

get_connected_peers_info

get_connected_peers_info() -> List[Dict[str, Any]]

Gets information about currently connected peers from the Go library.

Returns:

Type	Description
List[Dict[str, Any]]	A list of dictionaries, each representing a connected peer with
List[Dict[str, Any]]	keys like 'addr_info' (containing 'ID', 'Addrs'), 'connected_at', 'direction', and 'misc'.

Raises:

Type	Description
P2PError	If fetching connected peers fails.

Source code in unaiverse/networking/p2p/p2p.py

def get_connected_peers_info(self) -> List[Dict[str, Any]]:
    """
    Gets information about currently connected peers from the Go library.

    Returns:
        A list of dictionaries, each representing a connected peer with
        keys like 'addr_info' (containing 'ID', 'Addrs'), 'connected_at', 'direction', and 'misc'.

    Raises:
        P2PError: If fetching connected peers fails.
    """

    # Logger.info("ℹ️ Fetching connected peers info...") # Can be noisy
    try:

        # GetConnectedPeers takes no arguments in Go
        result_ptr = P2P.libp2p.GetConnectedPeers(P2P._type_interface.to_go_int(self._instance))
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error("Failed to get connected peers, received null result.")
            raise P2PError("Failed to get connected peers, received null result.")
        if result.get('state') == "Error":
            logger.error(f"Failed to get connected peers: {result.get('message', 'Unknown Go error')}")
            raise P2PError(f"Failed to get connected peers: {result.get('message', 'Unknown Go error')}")

        peers_list = result.get('message', [])

        # Update internal map (optional)
        # logger.info(f"  Connected peers count: {len(peers_list)}") # Can be noisy
        return peers_list

    except Exception as e:

        # Avoid crashing the polling thread, just log the error
        logger.error(f"❌ Error fetching connected peers info: {e}")

        # Optionally raise P2PError(f"Failed to get connected peers info") from e if called directly
        return []  # Return empty list on error during polling

get_rendezvous_peers_info

get_rendezvous_peers_info() -> Dict[str, Any] | None

Gets the full rendezvous state from the Go library, including peers and metadata.

Returns:

Type	Description
Dict[str, Any] | None	A dictionary representing the RendezvousState (containing 'peers', 'update_count', 'last_updated') if an update has been received.
Dict[str, Any] | None	None if no rendezvous topic is active or no updates have arrived yet.

Raises:

Type	Description
P2PError	If fetching the state fails in Go.

Source code in unaiverse/networking/p2p/p2p.py

def get_rendezvous_peers_info(self) -> Dict[str, Any] | None:
    """
    Gets the full rendezvous state from the Go library, including peers and metadata.

    Returns:
        - A dictionary representing the RendezvousState (containing 'peers',
          'update_count', 'last_updated') if an update has been received.
        - None if no rendezvous topic is active or no updates have arrived yet.

    Raises:
        P2PError: If fetching the state fails in Go.
    """
    try:
        result_ptr = P2P.libp2p.GetRendezvousPeers(P2P._type_interface.to_go_int(self._instance))
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error("Failed to get rendezvous peers, received null result.")
            raise P2PError("Failed to get rendezvous peers, received null result.")

        state = result.get('state')
        if state == "Empty":
            logger.debug(f"[Instance {self._instance}] GetRendezvousPeers: No rendezvous messages received yet.")
            return None  # Return None for the "empty" state
        elif state == "Error":
            error_msg = result.get('message', 'Unknown Go error')
            logger.error(f"Failed to get rendezvous peers: {error_msg}")
            raise P2PError(f"Failed to get rendezvous peers: {error_msg}")
        elif state == "Success":

            # The message payload is the full RendezvousState object
            rendezvous_state = result.get('message', {})
            return rendezvous_state
        else:
            logger.error(f"[Instance {self._instance}] GetRendezvousPeers: Received invalid state '{state}'.")
            raise P2PError(f"[Instance {self._instance}] GetRendezvousPeers: Received invalid state.")

    except Exception as e:

        # Avoid crashing the polling thread, just log the error
        logger.error(f"❌ Error fetching rendezvous peers info: {e}")

        # Optionally raise P2PError(f"Failed to get rendezvous peers info") from e if called directly
        return []  # Return empty list on error during polling

get_message_queue_length

get_message_queue_length() -> int

Gets the current number of messages in the incoming queue.

Returns:

Type	Description
int	The number of messages waiting.

Raises:

Type	Description
P2PError	If querying the length fails (should be rare).

Source code in unaiverse/networking/p2p/p2p.py

def get_message_queue_length(self) -> int:
    """
    Gets the current number of messages in the incoming queue.

    Returns:
        The number of messages waiting.

    Raises:
        P2PError: If querying the length fails (should be rare).
    """
    try:

        # Call Go function, returns C.int directly
        length_cint = P2P.libp2p.MessageQueueLength(P2P._type_interface.to_go_int(self._instance))
        length = P2P._type_interface.from_go_int(length_cint)

        # Print(f"  Current Message Queue Len: {length}") # Can be noisy
        return length
    except Exception as e:

        # Avoid crashing polling thread
        logger.error(f"❌ Error fetching message queue length: {e}")
        return -1  # Indicate error

close

close(close_all: bool = False) -> None

Gracefully shuts down the libp2p node and stops background threads.

Parameters:

Name	Type	Description	Default
close_all	bool	If True, closes all instances of the node. Default is False.	False

Source code in unaiverse/networking/p2p/p2p.py

def close(self, close_all: bool = False) -> None:
    """
    Gracefully shuts down the libp2p node and stops background threads.

    Args:
        close_all: If True, closes all instances of the node. Default is False.
    """
    logger.info("🛑 Closing node...")

    # 1. Signal background threads to stop
    logger.info("  - Stopping background threads...")
    self._stop_event.set()

    # 2. Wait briefly for threads to finish (optional, they are daemons)
    # self._get_connected_peers_thread.join(timeout=2)
    # self._check_message_queue_thread.join(timeout=2)
    # print("  - Background threads signaled.")

    # 3. Call the Go CloseNode function
    try:
        if close_all:
            result_ptr = P2P.libp2p.CloseNode(P2P._type_interface.to_go_int(-1))
        else:
            result_ptr = P2P.libp2p.CloseNode(P2P._type_interface.to_go_int(self._instance))
        result = P2P._type_interface.from_go_ptr_to_json(result_ptr)

        if result is None:
            logger.error("Node closure failed: received null result.")
            raise P2PError("Node closure failed: received null result.")
        if result.get('state') == "Error":
            logger.error(f"Node closure failed: {result.get('message', 'Unknown Go error')}")
            raise P2PError(f"Node closure failed: {result.get('message', 'Unknown Go error')}")

        close_msg = f"Node closed successfully ({'all instances' if close_all else f'instance {str(self._instance)}'})."
        logger.info(f"✅ {close_msg}")

    except Exception as e:
        logger.error(f"❌ Error closing node: {e}")
        raise P2PError(f"Error closing node: {e}") from e

    # 4. Clear internal state
    self._peer_id = None
    self._address_cache = None
    self._address_cache_time = time.monotonic()
    self._peer_map = {}
    with P2P._instance_lock:
        if close_all:

            # Also apply the lock here and use the corrected logic
            P2P._instance_ids = [False] * P2P._MAX_INSTANCES
            logger.info("🐍 All instance slots have been marked as free.")
        else:
            if self._instance != -1:  # Ensure instance was set
                P2P._instance_ids[self._instance] = False
                logger.info(f"🐍 Instance slot {self._instance} has been marked as free.")

    logger.info("🐍 Python P2P object state cleared.")

    return close_msg

P2PError

Bases: Exception

Custom exception class for P2P library errors.

TypeInterface

TypeInterface(libp2p_instance: GoLibP2P)

Helper class for converting between Python types and Go types using ctypes.

Methods:

Name	Description
to_go_string	Converts a Python string to a UTF-8 encoded Python 'bytes' object.
from_go_string	Converts a C char pointer (Go string) to a Python string.
to_go_int	Converts a Python integer to a Go-compatible ctypes integer.
from_go_int	Converts a ctypes.c_int from Go to a Python integer.
to_go_float	Converts a Python float to a Go-compatible ctypes float.
from_go_float	Converts a ctypes.c_float from Go to a Python float.
to_go_bool	Converts a Python boolean to a Go-compatible integer (1 if True, 0 if False).
from_go_bool	Converts a Go-compatible integer (ctypes.c_int) to a Python boolean.
to_go_bytes	Converts a Python bytes object to a Go-compatible C char pointer.
from_go_bytes	Converts a Go pointer representing a byte array to a Python bytes object.
from_go_ptr_to_json	Converts a C void* pointer (returned by Go as int) pointing to a
to_go_json	Encodes a Python object to a JSON string, returning a UTF-8 encoded
from_go_string_to_list	Decodes a JSON-encoded list from a Go C char pointer into a Python list.

Source code in unaiverse/networking/p2p/lib_types.py

def __init__(self, libp2p_instance: GoLibP2P):
    self.__freed_pointers: set[int] = set()  # Track freed pointers to prevent double-free errors
    self.__freed_pointers_lock: Any = Lock()  # A threading lock
    self.libp2p: GoLibP2P = libp2p_instance  # Store the shared library object instance

Methods:

to_go_string

to_go_string(s: str) -> bytes

Converts a Python string to a UTF-8 encoded Python 'bytes' object.

This 'bytes' object is suitable for direct use with ctypes when passing to a C function expecting a 'char*' (ctypes.c_char_p), as ctypes will automatically pass a pointer to the byte string's data.

Parameters:

Name	Type	Description	Default
s	str	The Python string.	required

Returns:

Type	Description
bytes	A Python 'bytes' object containing the UTF-8 encoded string.

Source code in unaiverse/networking/p2p/lib_types.py

def to_go_string(self, s: str) -> bytes:
    """
    Converts a Python string to a UTF-8 encoded Python 'bytes' object.

    This 'bytes' object is suitable for direct use with ctypes when passing
    to a C function expecting a 'char*' (ctypes.c_char_p), as ctypes
    will automatically pass a pointer to the byte string's data.

    Args:
        s: The Python string.

    Returns:
        A Python 'bytes' object containing the UTF-8 encoded string.
    """
    if s is None:
        s = ""
    return s.encode("utf-8")

from_go_string

from_go_string(cstr: bytes) -> str

Converts a C char pointer (Go string) to a Python string.

Parameters:

Name	Type	Description	Default
cstr	bytes	The C char pointer.	required

Returns:

Type	Description
str	The decoded Python string.

Source code in unaiverse/networking/p2p/lib_types.py

def from_go_string(self, cstr: bytes) -> str:
    """
    Converts a C char pointer (Go string) to a Python string.

    Args:
        cstr: The C char pointer.

    Returns:
        The decoded Python string.
    """
    if not cstr:
        return ""
    return cstr.decode("utf-8")

to_go_int

to_go_int(i: int) -> c_int

Converts a Python integer to a Go-compatible ctypes integer.

Parameters:

Name	Type	Description	Default
i	int	The Python integer.	required

Returns:

Type	Description
c_int	A ctypes.c_int equivalent.

Source code in unaiverse/networking/p2p/lib_types.py

def to_go_int(self, i: int) -> ctypes.c_int:
    """
    Converts a Python integer to a Go-compatible ctypes integer.

    Args:
        i: The Python integer.

    Returns:
        A ctypes.c_int equivalent.
    """
    return ctypes.c_int(i)

from_go_int

from_go_int(val: c_int) -> int

Converts a ctypes.c_int from Go to a Python integer.

Parameters:

Name	Type	Description	Default
val	c_int	The ctypes.c_int value.	required

Returns:

Type	Description
int	The corresponding Python integer.

Source code in unaiverse/networking/p2p/lib_types.py

def from_go_int(self, val: ctypes.c_int) -> int:
    """
    Converts a ctypes.c_int from Go to a Python integer.

    Args:
        val: The ctypes.c_int value.

    Returns:
        The corresponding Python integer.
    """
    return int(val)

to_go_float

to_go_float(f: float) -> c_float

Converts a Python float to a Go-compatible ctypes float.

Parameters:

Name	Type	Description	Default
f	float	The Python float.	required

Returns:

Type	Description
c_float	A ctypes.c_float equivalent.

Source code in unaiverse/networking/p2p/lib_types.py

def to_go_float(self, f: float) -> ctypes.c_float:
    """
    Converts a Python float to a Go-compatible ctypes float.

    Args:
        f: The Python float.

    Returns:
        A ctypes.c_float equivalent.
    """
    return ctypes.c_float(f)

from_go_float

from_go_float(val: c_float) -> float

Converts a ctypes.c_float from Go to a Python float.

Parameters:

Name	Type	Description	Default
val	c_float	The ctypes.c_float value.	required

Returns:

Type	Description
float	The corresponding Python float.

Source code in unaiverse/networking/p2p/lib_types.py

def from_go_float(self, val: ctypes.c_float) -> float:
    """
    Converts a ctypes.c_float from Go to a Python float.

    Args:
        val: The ctypes.c_float value.

    Returns:
        The corresponding Python float.
    """
    return float(val)

to_go_bool

to_go_bool(b: bool) -> c_int

Converts a Python boolean to a Go-compatible integer (1 if True, 0 if False).

Parameters:

Name	Type	Description	Default
b	bool	The Python boolean.	required

Returns:

Type	Description
c_int	A ctypes.c_int (1 or 0).

Source code in unaiverse/networking/p2p/lib_types.py

def to_go_bool(self, b: bool) -> ctypes.c_int:
    """
    Converts a Python boolean to a Go-compatible integer (1 if True, 0 if False).

    Args:
        b: The Python boolean.

    Returns:
        A ctypes.c_int (1 or 0).
    """
    return ctypes.c_int(1 if b else 0)

from_go_bool

from_go_bool(val: c_int) -> bool

Converts a Go-compatible integer (ctypes.c_int) to a Python boolean.

Parameters:

Name	Type	Description	Default
val	c_int	The ctypes.c_int value.	required

Returns:

Type	Description
bool	True if the value equals 1, False otherwise.

Source code in unaiverse/networking/p2p/lib_types.py

def from_go_bool(self, val: ctypes.c_int) -> bool:
    """
    Converts a Go-compatible integer (ctypes.c_int) to a Python boolean.

    Args:
        val: The ctypes.c_int value.

    Returns:
        True if the value equals 1, False otherwise.
    """
    return val == 1

to_go_bytes

to_go_bytes(b: bytes) -> c_char_p

Converts a Python bytes object to a Go-compatible C char pointer.

Parameters:

Name	Type	Description	Default
b	bytes	The Python bytes.	required

Returns:

Type	Description
c_char_p	A ctypes.c_char_p pointing to the byte data.

Source code in unaiverse/networking/p2p/lib_types.py

def to_go_bytes(self, b: bytes) -> ctypes.c_char_p:
    """
    Converts a Python bytes object to a Go-compatible C char pointer.

    Args:
        b: The Python bytes.

    Returns:
        A ctypes.c_char_p pointing to the byte data.
    """
    if b is None:
        b = b""
    buf = ctypes.create_string_buffer(b, len(b))
    return ctypes.cast(buf, ctypes.c_char_p)

from_go_bytes

from_go_bytes(cptr: c_char_p, length: int) -> bytes

Converts a Go pointer representing a byte array to a Python bytes object.

Parameters:

Name	Type	Description	Default
cptr	c_char_p	The C pointer to the byte array.	required
length	int	The number of bytes to read.	required

Returns:

Type	Description
bytes	A Python bytes object containing the read data.

Source code in unaiverse/networking/p2p/lib_types.py

def from_go_bytes(self, cptr: ctypes.c_char_p, length: int) -> bytes:
    """
    Converts a Go pointer representing a byte array to a Python bytes object.

    Args:
        cptr: The C pointer to the byte array.
        length: The number of bytes to read.

    Returns:
        A Python bytes object containing the read data.
    """
    if not cptr or length <= 0:
        return bytes()
    return ctypes.string_at(cptr, length)

from_go_ptr_to_json

from_go_ptr_to_json(c_void_ptr_val: int) -> Any

Converts a C void* pointer (returned by Go as int) pointing to a null-terminated C string containing JSON into a Python object.

It reads the string, parses it as JSON, and crucially frees the C memory using the provided FreeString function from the Go library.

Parameters:

Name	Type	Description	Default
c_void_ptr_val	int	The integer value representing the C pointer address.	required

Returns:

Type	Description
Any	The parsed Python object from the JSON string.

Raises:

Type	Description
GoLibError	If the pointer is NULL, reading/decoding fails, or JSON parsing fails.
TypeError	When go_lib is not a valid ctypes library object.

Source code in unaiverse/networking/p2p/lib_types.py

def from_go_ptr_to_json(self, c_void_ptr_val: int) -> Any:
    """
    Converts a C void* pointer (returned by Go as int) pointing to a
    null-terminated C string containing JSON into a Python object.

    It reads the string, parses it as JSON, and crucially frees the C memory
    using the provided FreeString function from the Go library.

    Args:
        c_void_ptr_val: The integer value representing the C pointer address.

    Returns:
        The parsed Python object from the JSON string.

    Raises:
        GoLibError: If the pointer is NULL, reading/decoding fails, or
                    JSON parsing fails.
        TypeError: When go_lib is not a valid ctypes library object.
    """

    json_string: Optional[str] = None  # To store the string for error reporting

    if not c_void_ptr_val:  # Check if the address is NULL (0)
        raise print("Received a NULL pointer from Go function")

    try:

        # --- Double-Free Check (Before Reading/Casting) ---
        self.__freed_pointers_lock.acquire()  # Acquire lock if using threading
        if c_void_ptr_val in self.__freed_pointers:

            # This indicates a serious logic error elsewhere - the pointer
            # was already freed but somehow passed here again.
            logger.warning(f"🔥🔥🔥 ATTEMPT TO PROCESS ALREADY FREED POINTER {hex(c_void_ptr_val)}! 🔥🔥🔥")

            # Raising an error is safer than trying to read potentially invalid memory.
            logger.error(f"Attempt to process pointer {hex(c_void_ptr_val)} which was already freed",
                         pointer_val=c_void_ptr_val)
            raise Exception(f"Attempt to process pointer {hex(c_void_ptr_val)} which was already freed",
                            pointer_val=c_void_ptr_val)
        self.__freed_pointers_lock.release()  # Release lock if using threading

        # --- Cast void* to c_char_p and Read String ---
        try:

            # Perform the cast only when needed for reading
            c_char_ptr_for_read = ctypes.cast(c_void_ptr_val, ctypes.c_char_p)
            raw_bytes = ctypes.string_at(c_char_ptr_for_read)
            json_string = raw_bytes.decode('utf-8')

            # Logger.debug(f"Read string (len={len(json_string)}) from pointer {hex(c_void_ptr_val)}: %.100s...", json_string)
        except (ctypes.ArgumentError, ValueError, UnicodeDecodeError) as read_err:
            logger.error(f"Failed to read/decode string from pointer {hex(c_void_ptr_val)}: {read_err}", exc_info=False)

            # Even if reading fails, the pointer itself *might* still be valid C memory
            # that Go expects us to free. We will proceed to free it in finally.
            raise Exception(f"Failed to read string from pointer {hex(c_void_ptr_val)}: {read_err}",
                            pointer_val=c_void_ptr_val) from read_err
        except Exception as unexpected_read_err:  # Catch other potential ctypes issues
            logger.error(f"Unexpected error reading C string from pointer {hex(c_void_ptr_val)}: {unexpected_read_err}", exc_info=True)
            raise Exception(f"Unexpected error reading C string from pointer {hex(c_void_ptr_val)}: {unexpected_read_err}",
                            pointer_val=c_void_ptr_val) from unexpected_read_err

        # --- Check for Empty String ---

        # --- Parse JSON ---
        try:

            # Now that we have the string, parse it
            logger.debug(f"Parsing JSON from string: {json_string}")
            parsed_data = json.loads(json_string)
            logger.debug(f"Parsed JSON data: {parsed_data}")

            # Logger.debug(f"Successfully parsed JSON from pointer {hex(c_void_ptr_val)}")
            return parsed_data  # Return the parsed Python object

        except json.JSONDecodeError as json_err:
            logger.error(f"Failed to decode JSON from pointer {hex(c_void_ptr_val)}: {json_err}", exc_info=False)

            # Again, the pointer is likely valid C memory, but the content is bad.
            # Let the block handle freeing.
            raise Exception(f"Failed to decode JSON from pointer {hex(c_void_ptr_val)}: {json_err}",
                            pointer_val=c_void_ptr_val) from json_err

    finally:

        # --- CRITICAL: Free C Memory ---
        # This block executes even if errors occurred during read/parse,
        # ensuring we attempt to free any non-NULL pointer received from Go.
        with self.__freed_pointers_lock:
            if c_void_ptr_val:
                logger.info(f"🐍 FINALLY: Freeing pointer {hex(c_void_ptr_val)}...")
                if c_void_ptr_val in self.__freed_pointers:

                    # This check is technically redundant if the initial check worked,
                    # but provides an extra safety layer in case of concurrency issues
                    # (if freed_pointers is shared without locks - which it shouldn't be).
                    logger.warning(f"🔥🔥🔥 DOUBLE FREE DETECTED in finally block for {hex(c_void_ptr_val)}! Skipping FreeString call again. 🔥🔥🔥")
                else:

                    # Add before calling free
                    try:
                        self.libp2p.FreeString(c_void_ptr_val)  # Pass the original void* value
                        logger.info(f"✅ FINALLY: FreeString successful for {hex(c_void_ptr_val)}.")
                    except Exception as free_err:

                        # Log if FreeString fails, but don't raise from finally
                        # as it might hide the original error.
                        logger.critical(f"🚨 FAILED TO FREE C MEMORY for pointer {hex(c_void_ptr_val)} via FreeString: {free_err}", exc_info=True)

to_go_json

to_go_json(data: Any) -> bytes

Encodes a Python object to a JSON string, returning a UTF-8 encoded Python 'bytes' object.

This 'bytes' object is suitable for direct use with ctypes when passing to a C function expecting a 'char*' (ctypes.c_char_p).

Parameters:

Name	Type	Description	Default
data	Any	The Python object (e.g., dict, list) to encode.	required

Returns:

Type	Description
bytes	A Python 'bytes' object containing the JSON string, UTF-8 encoded.

Source code in unaiverse/networking/p2p/lib_types.py

def to_go_json(self, data: Any) -> bytes:
    """
Encodes a Python object to a JSON string, returning a UTF-8 encoded
Python 'bytes' object.

This 'bytes' object is suitable for direct use with ctypes when passing
to a C function expecting a 'char*' (ctypes.c_char_p).

Args:
    data: The Python object (e.g., dict, list) to encode.

Returns:
    A Python 'bytes' object containing the JSON string, UTF-8 encoded.
"""
    json_str = json.dumps(data)
    return self.to_go_string(json_str)

from_go_string_to_list

from_go_string_to_list(cstr: c_char_p) -> List[Any]

Decodes a JSON-encoded list from a Go C char pointer into a Python list.

Parameters:

Name	Type	Description	Default
cstr	c_char_p	The Go string (C char pointer) containing a JSON list.	required

Returns:

Type	Description
List[Any]	A Python list representing the JSON data.

Source code in unaiverse/networking/p2p/lib_types.py

def from_go_string_to_list(self, cstr: ctypes.c_char_p) -> List[Any]:
    """
    Decodes a JSON-encoded list from a Go C char pointer into a Python list.

    Args:
        cstr: The Go string (C char pointer) containing a JSON list.

    Returns:
        A Python list representing the JSON data.
    """
    s = self.from_go_string(cstr)

    return json.loads(s)