nexalware-simulate

Act as a real Nexalware device, or a master orchestrating sub-devices, from a plain Python process. No embedded firmware, no physical board. The wire protocol underneath is plain MQTT with username/password auth, identical to what ESP32/MicroPython firmware speaks, this package just hides that behind a small set of methods.

Two very different things this is for:

• Simulating a device you haven't built yet — a circuit designed in a simulator (e.g. Proteus), with real Arduino sketch code implementing the Sub-Device Contract over Serial, bridged into your PC over a real or virtual COM port. Prove a project works, or run classroom demos, before anyone buys or solders a physical board.
• Running a genuine production master from a PC — a PC is strictly more capable than an ESP32, and nothing about the platform requires embedded hardware specifically, this is just the SDK for that.

Not what this is for: calling the REST API (device registration, telemetry history, schedules, etc.), that's nexalware. This package is specifically the MQTT device-connection side, kept separate so installing the REST SDK never pulls in a persistent MQTT client you don't need.

Install

pip install nexalware-simulate
# only if you're using SerialTransport (the Proteus/COM-port workflow):
pip install nexalware-simulate[serial]
# only if you're using WebSocketTransport:
pip install nexalware-simulate[websocket]
# only if you're using the nexalware-serial-bridge CLI (needs both):
pip install nexalware-simulate[serial-bridge]

Quickstart — a single simulated device

from nexalware_simulate import NexalwareDevice

device = NexalwareDevice(
    device_id="dev_a1b2c3",
    mqtt_username="d_a1b2c3d4",
    mqtt_password="your-device-password",
)

def on_command(cmd, params):
    print("Nexalware sent:", cmd)
    device.publish_status(relay="ON" if cmd == "ON" else "OFF")

device.on_command = on_command

device.connect()
device.start_heartbeat()  # keeps the device "online" without you managing a timer

Get mqtt_username/mqtt_password from the dashboard's Credentials tab for a device you've registered, the exact same credentials the MicroPython/Arduino references use.

Quickstart — a master with simulated sub-devices (Proteus)

from nexalware_simulate import MasterDevice
from nexalware_simulate.transports.serial import SerialTransport

master = MasterDevice(
    device_id="dev_master1",
    mqtt_username="d_master1x",
    mqtt_password="your-master-password",
)
master.connect()

# Bridges Proteus's COMPIM-connected COM port straight into the master -
# every message your Arduino sketch sends over Serial becomes a tracked
# sub-device, automatically.
transport = SerialTransport("COM3", baud_rate=9600)
transport.attach(master)

master.start_heartbeat()

That's the whole PC side. See the full Proteus walkthrough for the circuit + Arduino sketch side.

NexalwareDevice(device_id, mqtt_username, mqtt_password, mqtt_host=..., mqtt_port=...)

Param	Type	Required	Meaning
device_id	str	yes	This device's public id, e.g. "dev_a1b2c3".
mqtt_username	str	yes	From the dashboard's Credentials tab.
mqtt_password	str	yes	From the same tab - shown once, generate new credentials if lost.
mqtt_host	str	no	Override for a self-hosted deployment. Defaults to "mqtt.nexalware.com".
mqtt_port	int	no	Override for a self-hosted deployment. Defaults to 1883.

MasterDevice takes the exact same arguments - it's a NexalwareDevice with sub-device orchestration layered on top.

NexalwareDevice methods

connect(timeout=10.0)

Connects over MQTT (on a background network thread) and subscribes to this device's command topic. Blocks until the subscription is confirmed or timeout seconds elapse.

disconnect()

Stops the heartbeat (if running) and closes the connection cleanly.

publish_status(relay=None, state=None, telemetry=None, uptime=None)

Merges the given fields into the last published status and publishes it. Only pass what changed - device_id/ts are filled in automatically, and anything you published before is preserved unless you overwrite it.

device.publish_status(relay="ON")
device.publish_status(state={"temperature": 21.5}, telemetry=[{"metric": "temperature", "value": 21.5, "unit": "C"}])

start_heartbeat(interval_seconds=20.0)

Republishes the last known status on a timer, so the device stays "online" - the backend's offline detection is a heartbeat timeout (35s by default), not a connection check.

stop_heartbeat()

Stops a heartbeat started with start_heartbeat. Called automatically by disconnect().

Callbacks (set as plain attributes)

• device.on_connected = lambda: ... — MQTT connection up, command subscription confirmed.
• device.on_disconnected = lambda: ... — connection dropped.
• device.on_command = lambda cmd, params: ... — a command arrived for this device itself.
• device.on_error = lambda err: ...

MasterDevice - everything above, plus:

receive_sub_device_message(channel_id, raw)

Feed this whatever line arrives from a sub-device.

• channel_id (str, required) — A stable identifier for the physical connection this line arrived on (e.g. the serial port's path). One connection = one sub-device, same assumption the reference ESP32 master makes: a sub-device's identify is the only message that carries its id, every later message on the same channel_id is assumed to be from the same sub-device.
• raw (str, required) — One line of raw JSON, exactly as the sub-device sent it, per the Sub-Device Contract.

publish_status(...)

Same as NexalwareDevice's, but sub_devices is filled in automatically from everything sub-devices have reported so far.

master.on_sub_device_send = lambda channel_id, message: ...

The master wants to send message (a Sub-Device Contract JSON string) down to the sub-device on channel_id. Your transport listens for this and actually writes the bytes out, this is the one piece MasterDevice can't do for you.

SerialTransport

The ready-made local transport for the Proteus/COMPIM workflow (or any real board over USB-serial). Runs its own background thread reading lines off the port.

from nexalware_simulate.transports.serial import SerialTransport

transport = SerialTransport("COM3", baud_rate=9600)
transport.attach(master)  # wires the port's read/write to the master automatically
transport.close()

• path (str, required) — The OS-level serial/COM port, e.g. "COM3" (Windows) or "/dev/ttyUSB0" (macOS/Linux).
• baud_rate (int, optional) — Must match your sub-device's Serial.begin(...). Defaults to 9600.

Requires pyserial (pip install nexalware-simulate[serial], or plain pip install pyserial) - it's an optional extra, not a hard dependency, so installing nexalware-simulate alone never requires it, and it isn't importable from the package's top level, only from nexalware_simulate.transports.serial explicitly. Same-machine only - a COM port (real or a com0com virtual pair) is an OS-local construct that can't cross a network. See WebSocketTransport below when your master and your sub-device (or its Proteus simulation) run on different machines.

WebSocketTransport

The network-capable transport: instead of opening a local port, the master hosts a WebSocket server that sub-devices connect in to, the same model the reference ESP32 master already uses, so real firmware built against the Sub-Device Contract works against a PC-hosted master too. Built on websockets.sync.server, a real thread-based server, no asyncio, matching the rest of this package staying a plain synchronous library. Handles any number of sub-devices connecting simultaneously, each connection becomes its own tracked sub-device automatically, this is what makes a classroom scenario (one master, N students each simulating their own sub-device) work out of the box.

from nexalware_simulate.transports.websocket import WebSocketTransport

transport = WebSocketTransport(port=8080, token="a-shared-secret")
transport.attach(master)  # starts a background server thread; blocks until listening
transport.close()

• port (int, required) — Which port to listen on for incoming sub-device connections.
• host (str, optional) — Defaults to all interfaces ("0.0.0.0"). Set to "127.0.0.1" to only accept connections from this same machine.
• token (str, optional) — Required by connecting clients as ?token=... in the connection URL. Omit it only for same-machine/trusted-network testing - without one, anything that can reach port can inject a fake sub-device into this master. Checked before the WebSocket handshake completes, via process_request, before any message from an unauthorized connection can ever reach receive_sub_device_message. If omitted, a warning is printed to stderr so an open, unauthenticated listener is never silent.

Requires websockets (pip install nexalware-simulate[websocket], or plain pip install websockets) - an optional extra, not a hard dependency, and not importable from the package's top level, only from nexalware_simulate.transports.websocket explicitly.

nexalware-serial-bridge CLI

Bridges a local serial port to a remote WebSocketTransport - the piece that turns "Proteus running on this machine" into "a sub-device reachable by a master running anywhere else," without a second com0com pair or a third-party tool. Purely mechanical (no Nexalware protocol knowledge), so it's also reusable for bridging any local serial device, simulated or real, to a remote master.

nexalware-serial-bridge --port COM3 --baud 9600 --url ws://master-host:8080 --token a-shared-secret

Flag	Meaning
--port	Local serial/COM port, e.g. COM3 or /dev/ttyUSB0. Same port your COMPIM/board is wired to.
--baud	Baud rate, must match the sketch. Defaults to 9600.
--url	The remote master's WebSocketTransport URL, e.g. ws://master-host:8080.
--token	The WebSocketTransport's token, if it has one. Omit only if the master was started without one.

Requires the serial-bridge extra (pip install nexalware-simulate[serial-bridge]), which pulls in both pyserial and websockets.

See Remote & Multi-Machine Simulation for the full walkthrough, including the classroom scenario and a networking (NAT/firewall/tunnel) heads-up.

Why isn't this an MCP tool?

nexalware-mcp's tools are request/response, an agent calls one and gets an answer. A device or master needs a long-held, continuously-listening MQTT connection, publishing and reacting to commands in real time, a fundamentally different shape than a stateless tool call. Use this package directly in a script/process instead.

Links

• Full docs & Proteus walkthrough
• Remote & Multi-Machine Simulation - WebSocketTransport, the serial-bridge CLI, the classroom scenario.
• Device Orchestration / The Sub-Device Contract
• SDK Reference (Python) - for the REST side (registering devices, reading telemetry history, schedules).

License

MIT