Skip to content

quickstart.md

Latest commit

 

History

History
247 lines (175 loc) · 5.78 KB

quickstart.md

File metadata and controls

247 lines (175 loc) · 5.78 KB
Raw

Quick Start

This guide will help you get started with unetpy in minutes.

Prerequisites

  1. Install unetpy: pip install unetpy
  2. Have a UnetStack simulator or modem running

Connecting to a Node

from unetpy import UnetSocket

# Create a socket connection to a UnetStack node
sock = UnetSocket("localhost", 1100)

# Get the local node address
print(f"Local address: {sock.getLocalAddress()}")

# Always close when done
sock.close()

Using Context Managers

The recommended way is to use a context manager for automatic cleanup:

from unetpy import UnetSocket

with UnetSocket("localhost", 1100) as sock:
    print(f"Local address: {sock.getLocalAddress()}")
# Socket is automatically closed here

Sending Data

Simple Send

from unetpy import UnetSocket, Protocol

with UnetSocket("localhost", 1100) as sock:
    # Send to node 31 using USER protocol
    success = sock.send([1, 2, 3], to=31, protocol=Protocol.USER)
    print(f"Send {'succeeded' if success else 'failed'}")

Using Connect for Default Destination

from unetpy import UnetSocket, Protocol

with UnetSocket("localhost", 1100) as sock:
    # Set default destination
    sock.connect(31, Protocol.USER)

    # Now send without specifying destination
    sock.send([1, 2, 3])
    sock.send([4, 5, 6])
    sock.send([7, 8, 9])

Using Socket-Level Metadata

from unetpy import Gateway, Protocol, UnetSocket

with UnetSocket("localhost", 1100) as sock:
    sock.connect(31, Protocol.USER)
    sock.setTTL(4)
    sock.setMailbox("STATUS")
    sock.setMimeType("application/json+auvstatus")
    sock.setRemoteRecipient("TOPSIDE")
    sock.setSendMode(Gateway.SEMI_BLOCKING)

    # MIME type / mailbox / remote recipient promote the request to RemoteMessageReq.
    sock.send('{"battery": 87}')

If you do not call setServiceProvider(), plain datagrams are sent through the normal transport/routing/link/physical/datagram stack. Remote-message traffic prefers the REMOTE service when it is available.

Receiving Data

Blocking Receive

from unetpy import UnetSocket, Protocol, DatagramNtf

with UnetSocket("localhost", 1100) as sock:
    # Bind to receive on USER protocol
    sock.bind(Protocol.USER)

    # Set timeout (in milliseconds)
    sock.setTimeout(5000)

    # Wait for datagram
    ntf = sock.receive()
    if isinstance(ntf, DatagramNtf):
        print(f"Received from {ntf.from_}: {ntf.data}")
    else:
        print("Timeout - no datagram received")

Non-blocking Receive

from unetpy import UnetSocket, Protocol

with UnetSocket("localhost", 1100) as sock:
    sock.bind(Protocol.USER)
    sock.setTimeout(0)  # Non-blocking

    ntf = sock.receive()
    if ntf is None:
        print("No datagram available")

Two-Way Communication

Here's a complete example with two nodes:

from unetpy import UnetSocket, Protocol, DatagramNtf
import threading

def receiver():
    with UnetSocket("localhost", 1102) as sock:
        sock.bind(Protocol.USER)
        sock.setTimeout(5000)
        ntf = sock.receive()
        if isinstance(ntf, DatagramNtf):
            print(f"Node B received: {ntf.data}")

def sender():
    with UnetSocket("localhost", 1101) as sock:
        sock.send([1, 2, 3], to=31, protocol=Protocol.USER)
        print("Node A sent data")

# Start receiver in background
recv_thread = threading.Thread(target=receiver)
recv_thread.start()

# Send data
sender()

# Wait for receiver
recv_thread.join()

Low-Level Access

For advanced use cases, you can access the underlying fjåge Gateway:

from unetpy import UnetSocket, Services

with UnetSocket("localhost", 1100) as sock:
    gw = sock.getGateway()

    # Get agents
    phy = gw.agentForService(Services.PHYSICAL)
    print(f"PHY agent: {phy.name}")
    print(f"MTU: {phy.MTU}")

    # Get node info
    node = sock.agent("node")
    print(f"Node name: {node.nodeName}")
    print(f"Node address: {node.address}")

Resolving Node Names

from unetpy import UnetSocket

with UnetSocket("localhost", 1100) as sock:
    # Resolve node name to address
    addr_a = sock.host("A")
    addr_b = sock.host("B")
    print(f"Node A address: {addr_a}")
    print(f"Node B address: {addr_b}")

Parameter Change Notifications

from unetpy import UnetSocket, Services

with UnetSocket("localhost", 1100) as sock:
    # Subscribe to parameter changes
    node = sock.agentForService(Services.NODE_INFO)
    sock.onParamChange(node, "address", lambda new_value: print(f"Node address changed to {new_value}"))

Coordinate Conversions

Convert between local (meters) and GPS coordinates:

from unetpy import to_gps, to_local

# Define origin point (latitude, longitude)
origin = (1.34286, 103.84109)

# Convert local coordinates to GPS
x, y = 100.0, 100.0  # meters from origin
lat, lon = to_gps(origin, x, y)
print(f"GPS: {lat}, {lon}")

# Convert GPS back to local
x_back, y_back = to_local(origin, lat, lon)
print(f"Local: {x_back}, {y_back}")

Logging and Debugging

This library uses the standard Python logging system.

By default, the library does not emit logs to the console.

To see logs, configure Python’s logging in your application:

import logging

# Show all logs on stdout
logging.basicConfig(level=logging.DEBUG)

# Or, enable only logs from this library
logging.getLogger("fjagepy").setLevel(logging.DEBUG)

For troubleshooting, you can also send logs to a file:

logging.basicConfig(filename="debug.log", level=logging.DEBUG)

Next Steps