Skip to content

Latest commit

 

History

History

Folders and files

NameName
Last commit message
Last commit date

parent directory

..
 
 
 
 
 
 
 
 
 
 
 
 
 
 

README.md

Standalone

Communication Config

To connect your device to Enapter Cloud, you first need to generate a communication config.

Note

The following instruction applies only to v3 sites. If you have a v1 site, follow this tutorial to generate your communication config.

Generate a communication config using Enapter CLI:

enapter3 device communication-config generate --device-id "$YOUR_DEVICE_ID" --protocol MQTTS | jq .config | base64 --wrap=0

Device Implementation

Here's a basic example:

# my_device.py

import asyncio
import enapter

async def main():
    await enapter.standalone.run(MyDevice())

class MyDevice(enapter.standalone.Device):
    async def run(self):
        while True:
            await self.send_telemetry({})
            await self.send_properties({"model": "0.0.1"})
            await asyncio.sleep(10)

if __name__ == "__main__":
    asyncio.run(main())
What's going on?

The most straightforward way to implement your own standalone device is:

  1. Subclass enapter.standalone.Device.
  2. Override async def run(self) -> None method to send telemetry and properties.
  3. Pass an instance of your device to enapter.standalone.run.

enapter.standalone.run will take care of loading the communication config and connecting your device to an upstream.

Running

Use the generated communication config to run your device:

export ENAPTER_STANDALONE_COMMUNICATION_CONFIG="$YOUR_COMMUNICATION_CONFIG"
python my_device.py

Handling Commands

enapter.standalone.Device dispatches incoming command execution requests to the corresponding methods on your subclass.

If this command is defined in the manifest:

configure_connection:
  display_name: Configure Connection
  group: connection
  arguments:
    ip_address:
      display_name: IP Address
      type: string
      required: true
    token:
      display_name: Bearer Authentication Token
      type: string
      required: false

Then you would implement the following command handler in your device class:

async def cmd_configure_connection(ip_address: str, token: str | None = None):
    ...

By default, the cmd_ prefix is used to locate command handlers.

Synchronous Code

Blocking (CPU-bound) code should not be called directly. For example, if a function performs a CPU-intensive calculation for 1 second, all concurrent asyncio Tasks and IO operations would be delayed by 1 second.

Instead, use asyncio.to_thread to offload such work to a thread pool:

await asyncio.to_thread(blocking_call())

Running In Docker

Here's an example of a simple Dockerfile:

FROM python:3.13-alpine

WORKDIR /app

RUN python -m venv .venv
COPY requirements.txt requirements.txt
RUN .venv/bin/pip install -r requirements.txt

COPY script.py script.py

CMD [".venv/bin/python", "script.py"]

Warning

If you are using Enapter Gateway and running Linux, connect your containers to the host network to enable mDNS resolution:

docker run --network host ...