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=0Here'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:
- Subclass
enapter.standalone.Device. - Override
async def run(self) -> Nonemethod to send telemetry and properties. - 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.
Use the generated communication config to run your device:
export ENAPTER_STANDALONE_COMMUNICATION_CONFIG="$YOUR_COMMUNICATION_CONFIG"
python my_device.pyenapter.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: falseThen 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.
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())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 ...