micropython-skills/network

# Network

Code templates for network communication on MicroPython devices.
WiFi and network operations are **Cautious tier** — inform the user before connecting.

WiFi credentials should never be hardcoded in files saved to the device.
Use variables or prompt the user for values.

## WiFi Station Mode (Connect to Router)

```python
import network, time, json
try:
    sta = network.WLAN(network.STA_IF)
    sta.active(True)

    ssid = "YOUR_SSID"       # Replace with actual value
    password = "YOUR_PASS"   # Replace with actual value
    sta.connect(ssid, password)

    timeout = 15
    start = time.time()
    while not sta.isconnected():
        if time.time() - start > timeout:
            print("ERROR:WiFi connection timed out after " + str(timeout) + "s")
            raise SystemExit
        time.sleep(0.5)

    ip, mask, gw, dns = sta.ifconfig()
    print("RESULT:" + json.dumps({
        "ip": ip, "mask": mask, "gateway": gw, "dns": dns,
    }))
except Exception as e:
    print("ERROR:" + str(e))
```

## WiFi Access Point Mode

Create a WiFi hotspot on the device:

```python
import network, json
try:
    ap = network.WLAN(network.AP_IF)
    ap.active(True)
    ap.config(essid="ESP32-AP", password="12345678", authmode=network.AUTH_WPA2_PSK)

    ip, mask, gw, dns = ap.ifconfig()
    print("RESULT:" + json.dumps({
        "mode": "AP",
        "ssid": "ESP32-AP",
        "ip": ip,
    }))
except Exception as e:
    print("ERROR:" + str(e))
```

## HTTP Client (urequests)

GET request:
```python
import urequests, json
try:
    r = urequests.get("http://httpbin.org/get")
    print("RESULT:" + json.dumps({
        "status": r.status_code,
        "body_preview": r.text[:200],
    }))
    r.close()
except Exception as e:
    print("ERROR:" + str(e))
```

POST with JSON:
```python
import urequests, json
try:
    data = {"sensor": "dht22", "temp": 23.5}
    r = urequests.post(
        "http://example.com/api/data",
        json=data,
        headers={"Content-Type": "application/json"}
    )
    print("RESULT:" + json.dumps({"status": r.status_code}))
    r.close()
except Exception as e:
    print("ERROR:" + str(e))
```

Note: `urequests` is included in most MicroPython firmware builds. For HTTPS, the device needs sufficient memory (ESP32 usually OK, ESP8266 may struggle).

## MQTT Publish / Subscribe

Using the built-in `umqtt.simple` module:

```python
from umqtt.simple import MQTTClient
import json, time
try:
    client = MQTTClient(
        "esp32_client",       # Client ID
        "broker.hivemq.com",  # Broker address
        port=1883,
    )
    client.connect()
    print("LOG:Connected to MQTT broker")

    # Publish
    topic = "test/esp32"
    payload = json.dumps({"temp": 23.5, "hum": 61})
    client.publish(topic, payload)

    print("RESULT:" + json.dumps({
        "action": "publish",
        "topic": topic,
        "payload": payload,
    }))

    client.disconnect()
except Exception as e:
    print("ERROR:" + str(e))
```

Subscribe with callback:
```python
from umqtt.simple import MQTTClient
import json, time
try:
    received = []

    def on_message(topic, msg):
        received.append({"topic": topic.decode(), "msg": msg.decode()})

    client = MQTTClient("esp32_sub", "broker.hivemq.com", port=1883)
    client.set_callback(on_message)
    client.connect()
    client.subscribe(b"test/commands")
    print("LOG:Subscribed, waiting for messages...")

    # Check for messages with timeout
    deadline = time.time() + 10
    while time.time() < deadline:
        client.check_msg()
        time.sleep(0.1)
        if received:
            break

    client.disconnect()
    print("RESULT:" + json.dumps({"received": received}))
except Exception as e:
    print("ERROR:" + str(e))
```

Note: If `umqtt` is not available, install it: `mpremote mip install umqtt.simple`

## BLE (Bluetooth Low Energy)

ESP32 BLE advertising example:

```python
import bluetooth, json, struct
try:
    ble = bluetooth.BLE()
    ble.active(True)
    name = "ESP32-BLE"

    # Encode advertising payload
    adv_data = bytearray()
    # Flags
    adv_data += struct.pack("BBB", 2, 0x01, 0x06)
    # Complete name
    name_bytes = name.encode()
    adv_data += struct.pack("BB", len(name_bytes) + 1, 0x09) + name_bytes

    ble.gap_advertise(100000, adv_data)  # Advertise every 100ms

    print("RESULT:" + json.dumps({
        "ble_active": True,
        "name": name,
        "advertising": True,
    }))
except Exception as e:
    print("ERROR:" + str(e))
```

BLE scan for nearby devices:
```python
import bluetooth, json, time
try:
    ble = bluetooth.BLE()
    ble.active(True)
    devices = []

    def scan_cb(event, data):
        if event == 5:  # _IRQ_SCAN_RESULT
            addr_type, addr, adv_type, rssi, adv_data = data
            addr_hex = ":".join("{:02x}".format(b) for b in addr)
            devices.append({"addr": addr_hex, "rssi": rssi})

    ble.irq(scan_cb)
    ble.gap_scan(5000, 30000, 30000)  # Scan for 5 seconds
    time.sleep(6)

    # Deduplicate by address
    seen = {}
    for d in devices:
        if d["addr"] not in seen or d["rssi"] > seen[d["addr"]]["rssi"]:
            seen[d["addr"]] = d

    print("RESULT:" + json.dumps({"ble_devices": list(seen.values()), "count": len(seen)}))
except Exception as e:
    print("ERROR:" + str(e))
```

## NTP Time Sync

Synchronize device clock from the internet:

```python
import ntptime, time, json
try:
    ntptime.settime()
    t = time.localtime()
    print("RESULT:" + json.dumps({
        "synced": True,
        "utc": "{:04d}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}Z".format(*t[:6]),
    }))
except Exception as e:
    print("ERROR:NTP sync failed: " + str(e))
```

Note: Device must be connected to WiFi first. NTP uses UDP port 123.

## WebSocket (Minimal Server)

Run a simple WebSocket echo server on the device — useful for real-time data streaming to a browser:

```python
import socket, json, hashlib, binascii
try:
    s = socket.socket()
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(("0.0.0.0", 8080))
    s.listen(1)
    print("LOG:WebSocket server on port 8080")
    print("RESULT:" + json.dumps({"server": "started", "port": 8080}))
    # Accept one connection for demo, then close
    # For production, use asyncio-based server
    s.close()
except Exception as e:
    print("ERROR:" + str(e))
```

For full WebSocket support, consider `micropython-async` or `microdot` framework.