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]

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.

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
• Device Orchestration / The Sub-Device Contract
• SDK Reference (Python) - for the REST side (registering devices, reading telemetry history, schedules).

License

MIT