tessl/pypi-eclipse-zenoh

The python API for Eclipse zenoh - a high-performance networking library providing pub/sub, store/query and compute framework for zero-overhead communication

—

Pending

Overview

Eval results

Files

Session Management

Name: tessl/pypi-eclipse-zenoh
Author: tessl

Session management provides the core functionality for establishing connections, managing configuration, and discovering network peers in the Zenoh ecosystem. The Session is the central hub for all Zenoh operations.

Capabilities

Session Creation and Configuration

Create and configure Zenoh sessions with various connection options.

class Config:
    """Main configuration structure for Zenoh"""
    
    def __init__(self):
        """Create a default configuration"""
    
    @staticmethod
    def from_env() -> Config:
        """Create configuration from environment variables"""
    
    @staticmethod
    def from_file(path: str) -> Config:
        """Load configuration from a JSON5 file"""
    
    @staticmethod
    def from_json5(config: str) -> Config:
        """Create configuration from JSON5 string"""
    
    def get_json(self, key: str) -> str:
        """Get configuration value as JSON string"""
    
    def insert_json5(self, key: str, value: str) -> Config:
        """Insert configuration key-value pair from JSON5"""

def open(config: Config = None) -> Session:
    """
    Open a zenoh Session.
    
    Parameters:
    - config: Configuration for the session. If None, uses default config.
    
    Returns:
    Session object for Zenoh operations
    """

Session Operations

Core session functionality for managing the connection lifecycle and retrieving session information.

class Session:
    """A zenoh session"""
    
    @property
    def info(self) -> SessionInfo:
        """Get session information"""
    
    def zid(self) -> ZenohId:
        """Get the session's ZenohId"""
    
    def is_closed(self) -> bool:
        """Check if session is closed"""
    
    def close(self) -> None:
        """Close the session and release resources"""
    
    def undeclare(self, entity) -> None:
        """Undeclare an entity (publisher, subscriber, etc.)"""
    
    def new_timestamp(self) -> Timestamp:
        """Create a new timestamp for this session"""
    
    def declare_keyexpr(self, key_expr: str) -> KeyExpr:
        """Declare a key expression to optimize repeated usage"""
    
    def declare_querier(self, key_expr, **kwargs) -> Querier:
        """Declare a querier for repeated queries on the same key expression"""
    
    def put(self, key_expr, payload, **kwargs) -> None:
        """Put data directly through the session"""
    
    def delete(self, key_expr, **kwargs) -> None:
        """Delete data directly through the session"""
    
    def get(self, selector, **kwargs):
        """Query data through the session"""
    
    @property
    def liveliness(self) -> Liveliness:
        """Access liveliness operations for this session"""

Network Discovery

Discover available Zenoh peers and routers on the network.

def scout(
    what: WhatAmIMatcher = None,
    timeout: float = None,
    handler = None
) -> Scout:
    """
    Scout for routers and/or peers.
    
    Parameters:
    - what: What to scout for (routers, peers, clients)
    - timeout: Scout timeout in seconds
    - handler: Handler for receiving Hello messages
    
    Returns:
    Scout object for receiving discovery results
    """

class Scout:
    """Scout handler for network discovery"""
    
    @property
    def handler(self):
        """Get the scout handler"""
    
    def stop(self) -> None:
        """Stop scouting"""
    
    def try_recv(self):
        """Try to receive a Hello message without blocking"""
    
    def recv(self):
        """Receive a Hello message (blocking)"""
    
    def __iter__(self):
        """Iterate over Hello messages"""

Session Information

Access information about the current session and connected peers.

class SessionInfo:
    """Session information"""
    
    def zid(self) -> ZenohId:
        """Get session ZenohId"""
    
    def routers_zid(self) -> list:
        """Get list of connected routers' ZenohIds"""
    
    def peers_zid(self) -> list:
        """Get list of connected peers' ZenohIds"""

class ZenohId:
    """Global unique peer identifier"""
    
    def __str__(self) -> str:
        """String representation of ZenohId"""

Discovery Messages

Handle network discovery responses.

class Hello:
    """Scout discovery message"""
    
    @property
    def whatami(self) -> WhatAmI:
        """What type of Zenoh node this is"""
    
    @property
    def zid(self) -> ZenohId:
        """ZenohId of the discovered node"""
    
    @property
    def locators(self) -> list:
        """List of locators for the node"""

Node Types

Identify and match different types of Zenoh nodes.

class WhatAmI:
    """Node type identification"""
    ROUTER = ...
    PEER = ...
    CLIENT = ...

class WhatAmIMatcher:
    """Node type matcher for scouting"""
    
    @staticmethod
    def empty() -> WhatAmIMatcher:
        """Empty matcher (matches nothing)"""
    
    @staticmethod
    def router() -> WhatAmIMatcher:
        """Match routers only"""
    
    @staticmethod
    def peer() -> WhatAmIMatcher:
        """Match peers only"""
    
    @staticmethod
    def client() -> WhatAmIMatcher:
        """Match clients only"""
    
    def is_empty(self) -> bool:
        """Check if matcher is empty"""
    
    def matches(self, what: WhatAmI) -> bool:
        """Check if matcher matches a node type"""

Timestamp Management

Create and manage timestamps for ordering and synchronization.

class Timestamp:
    """Timestamp for ordering and synchronization"""
    
    def __init__(self, time: datetime, id: TimestampId):
        """Create timestamp with time and identifier"""
    
    def get_time(self) -> datetime:
        """Get the datetime component"""
    
    def get_id(self) -> TimestampId:
        """Get the timestamp identifier"""
    
    def get_diff_duration(self, other: Timestamp) -> timedelta:
        """Get duration difference with another timestamp"""
    
    def to_string_rfc3339_lossy(self) -> str:
        """Convert to RFC3339 string representation"""
    
    @staticmethod
    def parse_rfc3339(rfc3839: str) -> Timestamp:
        """Parse RFC3339 timestamp string"""

class TimestampId:
    """Unique identifier component of timestamp"""
    
    def __init__(self, bytes: bytes):
        """Create from byte array"""
    
    def __bytes__(self) -> bytes:
        """Convert to bytes"""

Logging Configuration

Configure logging for Zenoh operations.

def try_init_log_from_env() -> bool:
    """
    Initialize logging from RUST_LOG environment variable.
    Returns True if successful, False otherwise.
    """

def init_log_from_env_or(level: str):
    """
    Initialize logging from environment or use provided level.
    
    Parameters:
    - level: Default log level if RUST_LOG not set
    """

Usage Examples

Basic Session Setup

import zenoh

# Default configuration
config = zenoh.Config()
session = zenoh.open(config)

print(f"Session ID: {session.zid()}")
session.close()

Configuration from File

import zenoh

# Load configuration from file
config = zenoh.Config.from_file("zenoh-config.json5")
session = zenoh.open(config)

# Session operations...
session.close()

Network Discovery

import zenoh

def handle_hello(hello):
    print(f"Discovered {hello.whatami}: {hello.zid}")

# Scout for all node types
scout = zenoh.scout(handler=handle_hello)

# Let it run for a while
import time
time.sleep(2)

scout.stop()

Session Information

import zenoh

session = zenoh.open()
info = session.info

print(f"Session ZID: {info.zid()}")
print(f"Connected routers: {len(info.routers_zid())}")
print(f"Connected peers: {len(info.peers_zid())}")

session.close()

Install with Tessl CLI