Python SDK¶

Name: zephyr ide
Author: Mylonics

The Python SDK provides both synchronous and asynchronous interfaces for message communication.

Installation¶

Generate with SDK:

python -m struct_frame messages.proto --build_py --py_path generated/ --sdk

Parser Usage¶

from struct_frame_parser import Parser, HeaderType, PayloadType
from messages_sf import Status

# Create parser
parser = Parser()

# Encode
frame = parser.encode_basic(msg_id=1, msg=Status(value=42).to_bytes())

# Decode
for byte in frame:
    result = parser.parse_byte(byte)
    if result.valid:
        msg = Status.from_bytes(result.msg_data)
        print(f"Status: {msg.value}")

Message Router¶

from struct_frame_sdk import MessageRouter
from messages_sf import Status

router = MessageRouter()

# Subscribe to messages
@router.subscribe(Status)
def handle_status(msg: Status):
    print(f"Status: {msg.value}")

# Process incoming data
router.process_byte(byte)

Transports¶

Serial¶

import serial
from struct_frame_sdk.transports import SerialTransport

transport = SerialTransport('/dev/ttyUSB0', 115200)
transport.connect()
transport.send(msg_id, data)

Socket¶

import socket
from struct_frame_sdk.transports import SocketTransport

transport = SocketTransport('192.168.1.100', 8080)
transport.connect()
transport.send(msg_id, data)

Async Support¶

import asyncio
from struct_frame_sdk.async_transports import AsyncSerialTransport

async def main():
    transport = AsyncSerialTransport('/dev/ttyUSB0', 115200)
    await transport.connect()
    await transport.send(msg_id, data)

asyncio.run(main())