1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172

# SPDX-FileCopyrightText: 2026 Tim Cocks for Adafruit Industries
# SPDX-License-Identifier: MIT
import json
from os import getenv

import board
import audiobusio
import digitalio
import pwmio
import supervisor
import wifi
import socketpool
import rtc

from adafruit_bme280 import basic as adafruit_bme280
from adafruit_io.adafruit_io import IO_MQTT
import adafruit_ntp
from adafruit_debouncer import Debouncer
from embodiment_message_handler import EmbodimentMessageHandler
import adafruit_veml7700
import adafruit_lis3dh
import adafruit_drv2605
import adafruit_connection_manager
import adafruit_minimqtt.adafruit_minimqtt as MQTT
import neopixel

# Get WiFi details and Adafruit IO keys, ensure these are setup in settings.toml
# (visit io.adafruit.com if you need to create an account, or if you need your Adafruit IO key.)
ssid = getenv("CIRCUITPY_WIFI_SSID")
password = getenv("CIRCUITPY_WIFI_PASSWORD")
aio_username = getenv("ADAFRUIT_AIO_USERNAME")
aio_key = getenv("ADAFRUIT_AIO_KEY")

### WiFi ###
if not wifi.radio.connected:
    print(f"Connecting to {ssid}")
    wifi.radio.connect(ssid, password)
    print(f"Connected to {ssid}!")

# update system time
pool = socketpool.SocketPool(wifi.radio)
ntp = adafruit_ntp.NTP(pool, tz_offset=0, cache_seconds=3600)
cur_time = rtc.RTC()
cur_time.datetime = ntp.datetime

### Initialize hardware components ###
bme280 = adafruit_bme280.Adafruit_BME280_I2C(board.I2C())
veml7700 = adafruit_veml7700.VEML7700(board.I2C())
mic = audiobusio.PDMIn(board.D6, board.D5, sample_rate=16000, bit_depth=16)
buzzer = pwmio.PWMOut(board.D9, variable_frequency=True)

lis3dh = adafruit_lis3dh.LIS3DH_I2C(board.I2C())
lis3dh.data_rate = adafruit_lis3dh.DATARATE_1344_HZ

rgb_strip = neopixel.NeoPixel(board.D10, 8, brightness=0.3, auto_write=True)

drv = adafruit_drv2605.DRV2605(board.I2C())
drv.sequence[0] = adafruit_drv2605.Effect(15)  # Set the effect on slot 0.

embodiment_config = {
    "sensors": [
        {"type": "temperature", "sensor": bme280, "units": "C"},
        {"type": "pressure", "sensor": bme280, "units": "hPa"},
        {"type": "humidity", "sensor": bme280, "units": "%"},
        {"type": "lux", "sensor": veml7700, "units": "lux", "property": "autolux"},
        {"type": "pdm_mic", "sensor": mic, "units": "normalized_rms"},
        {
            "type": "accelerometer",
            "sensor": lis3dh,
            "units": "G",
        },
    ],
    "buttons": {},
    "piezo_buzzer": buzzer,
    "vibration_driver": drv,
    "neopixels": rgb_strip,
    "display": supervisor.runtime.display,
}

pins = [(board.D0, "D0"), (board.D1, "D1"), (board.D2, "D2")]
for pin_i in range(len(pins)):
    pin = pins[pin_i]
    dio = digitalio.DigitalInOut(pin[0])
    dio.direction = digitalio.Direction.INPUT
    # Pins D1 and D2 use different PULL from pin D0
    if pin_i == 0:
        dio.pull = digitalio.Pull.UP
    else:
        dio.pull = digitalio.Pull.DOWN
    btn = Debouncer(dio)
    embodiment_config["buttons"][pin[1]] = btn

embodiment_message_handler = EmbodimentMessageHandler(embodiment_config)

display = supervisor.runtime.display
display.root_group = embodiment_message_handler.main_group


# Define callback functions for MQTT
def connected(client):
    print(
        "Connected to Adafruit IO!  Listening for embodiment.client-to-mcu messages..."
    )
    # Subscribe to changes on a feed named embodiment.client-to-mcu.
    client.subscribe("embodiment.client-to-mcu")


def subscribe(_, __, topic, granted_qos):
    # This method is called when the client subscribes to a new feed.
    print(f"Subscribed to {topic} with QOS level {granted_qos}")


def unsubscribe(_, __, topic, pid):
    # This method is called when the client unsubscribes from a feed.
    print(f"Unsubscribed from {topic} with PID {pid}")


def disconnected(_):
    # Disconnected function will be called when the client disconnects.
    print("Disconnected from Adafruit IO!")


def publish(_, userdata, topic, pid):
    # This method is called when the client publishes data to a feed.
    print(f"Published to {topic} with PID {pid}. userdata is None ? {userdata is None}")
    if userdata is not None:
        print("Published User data: ", end="")
        print(userdata)


def message(_, feed_id, payload):
    print(f"Feed {feed_id} received new value: {payload}")
    data = json.loads(payload)
    if "messages" in data:
        resp_obj = embodiment_message_handler.handle_messages(data["messages"])
        io.publish("embodiment.mcu-to-client", json.dumps(resp_obj))


# Create a socket pool and ssl_context
pool = adafruit_connection_manager.get_radio_socketpool(wifi.radio)
ssl_context = adafruit_connection_manager.get_radio_ssl_context(wifi.radio)

# Initialize a new MQTT Client object
mqtt_client = MQTT.MQTT(
    broker="io.adafruit.com",
    port=8883,
    username=aio_username,
    password=aio_key,
    socket_pool=pool,
    ssl_context=ssl_context,
    is_ssl=True,
)

# Initialize an Adafruit IO MQTT Client
io = IO_MQTT(mqtt_client, feed_history_enabled=False)

# Connect the callback methods defined above to Adafruit IO
io.on_connect = connected
io.on_disconnect = disconnected
io.on_subscribe = subscribe
io.on_unsubscribe = unsubscribe
io.on_message = message
io.on_publish = publish

print("Connecting to Adafruit IO...")
io.connect()

# Increase message size limit for history disabled feed
io._client.mqtt_msg = 1024 * 100  # pylint: disable=protected-access

while True:
    io.loop()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317

#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2026 Tim Cocks for Adafruit Industries
# SPDX-License-Identifier: MIT
"""CLI for sending commands to an MCU over Adafruit IO MQTT or HTTP and waiting for an ack."""

# pylint: disable=import-outside-toplevel, too-many-locals, too-many-return-statements
import argparse
import json
import os
import sys
import time
import uuid
from os import getenv

import requests

DEFAULT_TIMEOUT = 10  # sec
TX_FEED = "embodiment.client-to-mcu"
RX_FEED = "embodiment.mcu-to-client"
HTTP_ENDPOINT = "/embodiment_kit"


def parse_args():
    parser = argparse.ArgumentParser(
        description="Send a command to the embodiment MCU and wait for ack. Output response JSON.",
    )
    parser.add_argument(
        "command",
        nargs="?",
        help="Name of the command to send (e.g. show_squinch_face).",
    )
    parser.add_argument(
        "arguments",
        nargs="*",
        metavar="key=value",
        help="Optional arguments as key=value pairs (e.g. frequency=440 duration=0.5).",
    )
    parser.add_argument(
        "-t",
        "--timeout",
        type=float,
        default=None,
        help=f"Seconds to wait for ack (default: {DEFAULT_TIMEOUT}, or 22 for show_prompt).",
    )
    parser.add_argument(
        "-q",
        "--quiet",
        action="store_true",
        help="Suppress connection/status logging; only print the ack response.",
    )
    parser.add_argument(
        "--commandlist",
        action="store_true",
        help="Print available commands and their arguments, then exit.",
    )
    return parser.parse_args()


def parse_arguments(raw_args):
    result = {}
    for item in raw_args:
        if "=" not in item:
            raise argparse.ArgumentTypeError(
                f"Argument must be key=value format, got: {item!r}"
            )
        key, _, raw_value = item.partition("=")
        if raw_value.startswith("0x") or raw_value.startswith("0X"):
            try:
                value = int(raw_value, 16)
            except ValueError:
                value = raw_value
        else:
            try:
                value = json.loads(raw_value)
            except json.JSONDecodeError:
                value = raw_value.encode("raw_unicode_escape").decode("unicode_escape")
        result[key] = value
    return result


def build_payload(command_name, arguments, command_uuid):
    return {
        "messages": [
            {
                "metadata": {"type": "command"},
                "command": {
                    "name": command_name,
                    "arguments": arguments,
                    "uuid": command_uuid,
                },
            }
        ]
    }


def normalize_host_url(host):
    """Ensure the host has a scheme; default to http:// if missing."""
    host = host.strip().rstrip("/")
    if not host.startswith(("http://", "https://")):
        host = "http://" + host
    return host


def send_via_http(host, payload, command_uuid, timeout, log):
    """Send the command via HTTP POST to the embodiment kit device.

    Returns the ack message dict on success, or None on failure/timeout.
    """
    url = normalize_host_url(host) + HTTP_ENDPOINT
    log(f"POSTing to {url}...")

    try:
        response = requests.post(
            url,
            json=payload,
            timeout=timeout,
            headers={"Content-Type": "application/json"},
        )
    except requests.exceptions.Timeout:
        log(f"HTTP request timed out after {timeout}s")
        return None
    except requests.exceptions.RequestException as e:
        log(f"HTTP request failed: {e}")
        return None

    if response.status_code != 200:
        log(f"HTTP request returned status {response.status_code}: {response.text}")
        return None

    try:
        data = response.json()
    except ValueError as e:
        log(f"Could not decode response JSON: {e}")
        return None

    # The device responds with the same envelope format. Find the ack message
    # matching our command uuid.
    for msg in data.get("messages", []):
        metadata = msg.get("metadata", {})
        command = msg.get("command", {})
        if metadata.get("type") == "ack" and command.get("uuid") == command_uuid:
            return msg

    # If no ack-typed message is found but there's exactly one message, accept it
    # (some devices may not tag the response as ack).
    messages = data.get("messages", [])
    if len(messages) == 1:
        return messages[0]

    log("No matching ack found in HTTP response")
    return None


def send_via_aio(aio_username, aio_key, payload, command_uuid, timeout, log):
    """Send the command via Adafruit IO MQTT and wait for an ack.

    Returns the ack message dict on success, or None on timeout.
    """
    # Imported lazily so HTTP-only users don't need these packages installed.
    import adafruit_connection_manager
    import adafruit_minimqtt.adafruit_minimqtt as MQTT
    from adafruit_io.adafruit_io import IO_HTTP, IO_MQTT

    state = {"status": "waiting", "uuid": command_uuid, "response": None}

    radio = adafruit_connection_manager.CPythonNetwork()
    socket_pool = adafruit_connection_manager.get_radio_socketpool(radio)
    ssl_context = adafruit_connection_manager.get_radio_ssl_context(radio)

    # pylint: disable=unused-argument
    # The arguments aren't used, but omitting them causes exception from MQTT library
    def on_connect(client):
        client.subscribe(RX_FEED)

    def on_disconnect(client):
        pass

    def on_subscribe(client, userdata, topic, granted_qos):
        pass

    def on_unsubscribe(client, userdata, topic, pid):
        pass

    def on_message(client, feed_id, raw_payload):
        try:
            data = json.loads(raw_payload)
        except json.JSONDecodeError as e:
            log(f"Could not decode message payload: {e}")
            return

        for msg in data.get("messages", []):
            metadata = msg.get("metadata", {})
            command = msg.get("command", {})
            if metadata.get("type") == "ack" and command.get("uuid") == state["uuid"]:
                state["status"] = "received"
                state["response"] = msg
                return

    mqtt_client = MQTT.MQTT(
        broker="io.adafruit.com",
        port=8883,
        username=aio_username,
        password=aio_key,
        is_ssl=True,
        socket_pool=socket_pool,
        ssl_context=ssl_context,
    )

    io_rx = IO_MQTT(mqtt_client)
    io_rx.on_connect = on_connect
    io_rx.on_disconnect = on_disconnect
    io_rx.on_subscribe = on_subscribe
    io_rx.on_unsubscribe = on_unsubscribe
    io_rx.on_message = on_message

    io_rx.connect()

    io_tx = IO_HTTP(aio_username, aio_key, requests)
    tx_feed = io_tx.get_feed(TX_FEED)

    io_tx.send_data(tx_feed["key"], json.dumps(payload))

    start = time.monotonic()
    while time.monotonic() - start < timeout and state["status"] == "waiting":
        io_rx.loop()

    io_rx.disconnect()

    if state["status"] == "received":
        return state["response"]
    return None


def main():
    args = parse_args()

    if args.commandlist:
        help_path = os.path.join(os.path.dirname(__file__), "embodiment_cli_help.md")
        with open(help_path) as f:
            print(f.read())
        return 0

    if not args.command:
        print("ERROR: command is required.", file=sys.stderr)
        sys.exit(2)

    def log(*msg):
        if not args.quiet:
            print(*msg, file=sys.stderr)

    # Determine which transport to use. Prefer the local HTTP device if
    # EMBODIMENT_KIT_HOST is set; otherwise fall back to Adafruit IO.
    embodiment_host = getenv("EMBODIMENT_KIT_HOST")
    aio_username = getenv("ADAFRUIT_AIO_USERNAME")
    aio_key = getenv("ADAFRUIT_AIO_KEY")

    use_http = bool(embodiment_host)
    use_aio = bool(aio_username and aio_key)

    if not use_http and not use_aio:
        print(
            "ERROR: no transport configured. Set EMBODIMENT_KIT_HOST for direct HTTP, "
            "or set ADAFRUIT_AIO_USERNAME and ADAFRUIT_AIO_KEY for Adafruit IO.",
            file=sys.stderr,
        )
        sys.exit(2)

    command_uuid = str(uuid.uuid4())
    try:
        arguments = parse_arguments(args.arguments)
    except argparse.ArgumentTypeError as e:
        print(f"ERROR: {e}", file=sys.stderr)
        sys.exit(2)

    user_specified_timeout = args.timeout is not None
    if args.command == "show_prompt":
        timeout = args.timeout if user_specified_timeout else 22.0
        duration = arguments.get("timeout")
        if duration is not None:
            if user_specified_timeout and timeout <= duration:
                print(
                    f"ERROR: --timeout ({timeout}s) must be greater than"
                    + f" duration ({duration}s) for show_prompt.",
                    file=sys.stderr,
                )
                sys.exit(2)
            elif not user_specified_timeout and duration >= timeout:
                timeout = duration + 2
        args.timeout = timeout
    else:
        if not user_specified_timeout:
            args.timeout = DEFAULT_TIMEOUT

    payload = build_payload(args.command, arguments, command_uuid)

    if use_http:
        response = send_via_http(
            embodiment_host, payload, command_uuid, args.timeout, log
        )
    else:
        response = send_via_aio(
            aio_username, aio_key, payload, command_uuid, args.timeout, log
        )

    if response is not None:
        print(json.dumps(response))
        return 0

    print(
        f"ERROR: timed out after {args.timeout}s waiting for ack to uuid {command_uuid}",
        file=sys.stderr,
    )
    return 1


if __name__ == "__main__":
    sys.exit(main())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237

---
name: embodiment-kit
description: Interact with a physical embodiment kit microcontroller located in the user's room. Send commands to control a display (faces, messages, prompts), NeoPixel lights, a piezo buzzer, and a vibration motor — and to read sensors (temperature, humidity, pressure, ambient lux, microphone, accelerometer). Use to answer questions about the physical environment and to express status, get the human's attention, or communicate visually/audibly/haptically in the room.
metadata:
  version: "1.1"
  requires: ["python3", "either (ADAFRUIT_AIO_USERNAME + ADAFRUIT_AIO_KEY env vars) or (EMBODIMENT_KIT_HOST env var)"]
---

# Embodiment Kit Skill

This skill lets you read sensors from and drive actuators on a physical embodiment kit microcontroller located in the user's room. Every command round-trips and returns a JSON ack containing the current sensor readings and output state — and, for actuator commands, a `proof` block with before/after sensor readings that confirm the action had a physical effect.

## When to Use This Skill

Use this skill when you need to:

- **Sense the room**: answer questions about the current temperature, humidity, pressure, ambient light, sound level, or device orientation.
- **Express into the room**: signal status, completion, attention, or mood using the display, lights, buzzer, or vibration motor.
- **Interact with the human**: display a message they should read, or pose a multiple-choice question and wait for a button press.
- **Verify physical effects**: confirm an actuator actually fired by reading the `proof` block in the ack.

Choose actuators thoughtfully — these produce real light, sound, and motion in the human's room. Avoid firing them gratuitously or when the human may be sleeping/focused unless they've asked for it.

## Requirements

- Python virtual environment with `requests` installed.
- If using the Adafruit IO transport (see below), also need installed:
  - `adafruit-circuitpython-adafruitio`
  - `adafruit-circuitpython-connectionmanager`
  - `adafruit-circuitpython-minimqtt`
- One of the two transports below must be configured via environment variables.
- CLI script at `<skills_directory>/embodiment-kit/scripts/embodiment_cli.py`.

### Transports

The CLI supports two ways of reaching the device. Set one (the CLI prefers HTTP when both are present):

- **Direct HTTP (recommended for same-network use)**: set `EMBODIMENT_KIT_HOST` to the device's URL or hostname, e.g. `http://192.168.1.189:5000` or `embodiment.local:5000`. A scheme is optional; `http://` is assumed if missing. Port `5000` is default. The CLI POSTs the command JSON to `<host>/embodiment_kit` and reads the ack from the HTTP response — no MQTT broker or Adafruit account needed. This is the lowest-latency option and works fully offline from the public internet.
- **Adafruit IO MQTT (for remote / cross-network use)**: set both `ADAFRUIT_AIO_USERNAME` and `ADAFRUIT_AIO_KEY`. The CLI publishes the command on the `embodiment.client-to-mcu` feed and waits for the ack on `embodiment.mcu-to-client`. Use this when you cannot reach the device directly on the local network.

If neither transport's env vars are set, the CLI exits with an error before sending anything.

## How to Run a Command

General form:

```bash
python <skills_directory>/embodiment-kit/scripts/embodiment_cli.py <command> [key=value ...] [-t TIMEOUT] [-q]
```

- `key=value` arguments are parsed as JSON when possible. Integers like `0xff8800` are parsed as hex. Strings with spaces should be quoted at the shell level (`message="Hello world"`).
- `-q` / `--quiet` suppresses connection logging on stderr, leaving stdout as pure JSON suitable for piping into `jq` or `json.loads`.
- Default timeout is 10s (22s for `show_prompt`). Override with `-t SECONDS`.
- `--commandlist` prints the full command reference from `embodiment_cli_help.md`.

Every successful response is a single line of JSON on stdout, parseable directly:

```json
{
  "state": {
    "sensors": { "...": "current readings" },
    "outputs": { "neopixels": {}, "display": "..." }
  },
  "metadata": { "type": "ack", "proof": {} },
  "command": { "name": "...", "uuid": "...", "arguments": {} }
}
```

The `metadata.proof` field is only present on actuator commands. To consume the output programmatically, pipe through `jq` or read it with `json.loads`:

```bash
python embodiment_cli.py get_data -q | jq '.state.sensors.temperature'
python embodiment_cli.py get_data -q | python -c "import sys, json; print(json.load(sys.stdin)['state']['sensors']['temperature'])"
```

## Sensors (always returned in every ack)

Under `state.sensors`:

- `temperature` — degrees C
- `humidity` — percent
- `pressure` — hPa
- `lux` — ambient light
- `pdm_mic` — normalized RMS sound magnitude
- `accelerometer` — string `"x=<x>, y=<y>, z=<z> G"`, averaged over 10 samples. A device sitting flat reads roughly `z≈9.8`.

### `get_data`
Returns the standard ack with no side effects. Use to answer questions about the room's current state.

```bash
python embodiment_cli.py get_data
```

## Display Face Commands

### Preset Faces
- `show_happy_face`
- `show_sad_face`
- `show_angry_face`
- `show_confused_face`
- `show_sleepy_face`

Shows the specified preset face. Optional `background_color` (24-bit RGB int, default `0x88ddff`).

```bash
python embodiment_cli.py show_happy_face
python embodiment_cli.py show_happy_face background_color=0xffcc00
python embodiment_cli.py show_sleepy_face
python embodiment_cli.py show_sleepy_face background_color=0xffcc00
```

### `show_custom_face`
Build a face from layer choices. First call `list_custom_face_options` to see valid filenames for each layer (`eyebrows`, `eyes`, `blush`, `mouth`). Each layer is a `[filename, y_offset]` pair. The y_offset is relative to the center of the screen. Negative values move the sprite up, positive values move it down. The display height is only 135px, offset values should be within the range -50 to 50 in order to keep them visible.

```bash
python embodiment_cli.py list_custom_face_options
python embodiment_cli.py show_custom_face options='{"eyes":["round_eyes_happy.png",-20],"mouth":["mouth_smiling.png",60]}'
```

## Display Message Commands

### `show_message` — important formatting rules
Hides the face and displays text.

**Display constraints — the screen fits at most 5 rows of 19 characters per row.** Plan messages around this:

- Keep each line ≤ 19 characters. Longer lines will be cut off.
- Use at most 5 lines total. Additional lines will not render.
- Use `\n` (literal backslash-n in the shell argument) to break lines manually — do NOT rely on auto-wrapping.
- Prefer short, scannable phrasing over paragraphs. Abbreviate when needed.

```bash
python embodiment_cli.py show_message message="Build passed\nReady to deploy"
python embodiment_cli.py show_message message="Tests: 42/42 OK\nLint: clean\nCoverage: 87%"
```

Bad (too long per line, will be truncated):

```bash
# DON'T: 23 chars on line 1, no newlines — will overflow / wrap unpredictably
python embodiment_cli.py show_message message="The deployment finished successfully a moment ago"
```

### `show_prompt` — important formatting rules
Displays a message and waits for the human to press one of three hardware buttons (`D0`, `D1`, `D2`) or time out.

**Same display constraints as `show_message`: 5 rows × 19 chars.** Use them to your advantage by structuring the prompt as a question followed by labeled options on their own lines.

Recommended structure: question on line 1, then up to three options on the remaining lines, each labeled with the button index that selects it (`0)`, `1)`, `2)`):

```bash
python embodiment_cli.py show_prompt message="Deploy now?\n0) Yes\n1) No\n2) Later" timeout=30
python embodiment_cli.py show_prompt message="Mood?\n0) Good\n1) OK\n2) Tired"
```

Arguments:
- `message` (string, default `"Press a button"`).
- `timeout` (seconds, default `20`). If you set this, the CLI's own ack-timeout (`-t`) is auto-bumped to `timeout + 2`. If you override both, `-t` must be strictly greater than `timeout`.

The ack's `command.response.prompt_result` will be one of:
- `"btn D0 pressed"` → option 0 chosen
- `"btn D1 pressed"` → option 1 chosen
- `"btn D2 pressed"` → option 2 chosen
- `"timed out"` → no press before timeout

Map the button → option by the labels you wrote in the message. After a press the display briefly shows `"Thank you"`; on timeout it shows `"Prompt timed out"`.

Tips:
- Only three buttons exist — never offer more than three options.
- Keep option labels very short (e.g. `Yes`, `No`, `Later`) so the `N) label` line fits in 19 chars.
- If asking a yes/no question, you can use 2 options and leave the third button unused.
- Pick a `timeout` long enough that the human can read and decide, but not so long that the prompt blocks the conversation indefinitely (20–60s is usually right).

## Light Commands (NeoPixel strip)

### `lights_on`
Turns on the strip. `color` is a 24-bit RGB int (default magenta `0xff00ff`). `brightness` is 0.0–1.0 (default is the current brightness).

```bash
python embodiment_cli.py lights_on color=0x00ff00 brightness=0.4
```

Proof block: `idle_lux_level` vs `on_lux_level` (lux should rise).

### `lights_off`
Fills the strip with black.

```bash
python embodiment_cli.py lights_off
```

Proof block: `idle_lux_level` vs `off_lux_level` (lux should drop).

## Sound Command

### `play_tone`
Plays a tone on the piezo buzzer.

- `frequency` (Hz, default `440`)
- `duty_cycle` (default `2**15` ≈ 50%)
- `duration` (seconds, default `0.5`, **max `3`**)

```bash
python embodiment_cli.py play_tone frequency=523 duration=0.8
```

Proof block: `idle_sound_level` vs `playing_sound_level` (mic RMS should rise).

## Haptic Command

### `vibrate`
Runs the vibration motor for ~0.75 s. No arguments.

```bash
python embodiment_cli.py vibrate
```

Proof block: `idle_z_acceleration` vs `vibrating_z_acceleration` (Z-axis acceleration usually decreases during vibration).

## Patterns for Expressing Status

Some useful patterns for using actuators meaningfully:

- **Task complete, low-key**: `show_message message="Done"` or a brief `play_tone frequency=880 duration=0.2`.
- **Task complete, celebratory**: `show_happy_face` + `lights_on color=0x00ff00 brightness=0.5`, then `lights_off` a few seconds later.
- **Need attention**: `vibrate` + `show_message` describing what you need.
- **Error / warning**: `lights_on color=0xff0000 brightness=0.7` + `show_message message="Build failed\nsee terminal"`.
- **Ask the human a question**: `show_prompt` with a short question and 2–3 labeled options.
- **Ambient status indicator**: `lights_on` with a color encoding state (e.g. blue=working, green=done, red=blocked), turn off when state ends.

Always pair attention-grabbing actuators (sound, vibration, bright lights) with a display message that explains *why* you grabbed attention, so the human knows what to do next. Clean up after yourself — turn lights off when the signal is no longer relevant.

## Reference Files

- `embodiment_cli.py` — the CLI itself.
- `embodiment_cli_help.md` — the canonical command reference (also printable via `--commandlist`).
- `embodiment_message_handler.py`, `code_mcu_mqtt.py`, `code_mcu_httpserver.py` — MCU-side firmware (for context, not invoked from this skill).

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

# SPDX-FileCopyrightText: 2026 Tim Cocks for Adafruit Industries
# SPDX-License-Identifier: MIT
import json
from os import getenv

import board
import supervisor
import wifi
import socketpool
import rtc
import digitalio
import audiobusio
import pwmio

from adafruit_bme280 import basic as adafruit_bme280
from adafruit_debouncer import Debouncer
import adafruit_veml7700
import adafruit_lis3dh
import adafruit_drv2605
import neopixel
import adafruit_ntp
from embodiment_message_handler import EmbodimentMessageHandler
from adafruit_httpserver import Request, Response, Server, POST

# Get WiFi details and Adafruit IO keys, ensure these are setup in settings.toml
# (visit io.adafruit.com if you need to create an account, or if you need your Adafruit IO key.)
ssid = getenv("CIRCUITPY_WIFI_SSID")
password = getenv("CIRCUITPY_WIFI_PASSWORD")

### WiFi ###
if not wifi.radio.connected:
    print(f"Connecting to {ssid}")
    wifi.radio.connect(ssid, password)
    print(f"Connected to {ssid}!")

# update system time
pool = socketpool.SocketPool(wifi.radio)
ntp = adafruit_ntp.NTP(pool, tz_offset=0, cache_seconds=3600)
cur_time = rtc.RTC()
cur_time.datetime = ntp.datetime

### Initialize hardware components ###
bme280 = adafruit_bme280.Adafruit_BME280_I2C(board.I2C())
veml7700 = adafruit_veml7700.VEML7700(board.I2C())
mic = audiobusio.PDMIn(board.D6, board.D5, sample_rate=16000, bit_depth=16)
buzzer = pwmio.PWMOut(board.D9, variable_frequency=True)

lis3dh = adafruit_lis3dh.LIS3DH_I2C(board.I2C())
lis3dh.data_rate = adafruit_lis3dh.DATARATE_1344_HZ

rgb_strip = neopixel.NeoPixel(board.D10, 8, brightness=0.3, auto_write=True)

drv = adafruit_drv2605.DRV2605(board.I2C())
drv.sequence[0] = adafruit_drv2605.Effect(15)  # Set the effect on slot 0.

embodiment_config = {
    "sensors": [
        {"type": "temperature", "sensor": bme280, "units": "C"},
        {"type": "pressure", "sensor": bme280, "units": "hPa"},
        {"type": "humidity", "sensor": bme280, "units": "%"},
        {"type": "lux", "sensor": veml7700, "units": "lux", "property": "autolux"},
        {"type": "pdm_mic", "sensor": mic, "units": "normalized_rms"},
        {
            "type": "accelerometer",
            "sensor": lis3dh,
            "units": "G",
        },
    ],
    "buttons": {},
    "piezo_buzzer": buzzer,
    "vibration_driver": drv,
    "neopixels": rgb_strip,
    "display": supervisor.runtime.display,
}

pins = [(board.D0, "D0"), (board.D1, "D1"), (board.D2, "D2")]
for pin_i in range(len(pins)):
    pin = pins[pin_i]
    dio = digitalio.DigitalInOut(pin[0])
    dio.direction = digitalio.Direction.INPUT
    # Pins D1 and D2 use different PULL from pin D0
    if pin_i == 0:
        dio.pull = digitalio.Pull.UP
    else:
        dio.pull = digitalio.Pull.DOWN
    btn = Debouncer(dio)
    embodiment_config["buttons"][pin[1]] = btn

embodiment_message_handler = EmbodimentMessageHandler(embodiment_config)

display = supervisor.runtime.display
display.root_group = embodiment_message_handler.main_group

# Set up HTTPServer
server = Server(pool, "/static", debug=True)
server.start(str(wifi.radio.ipv4_address), 5000)


@server.route("/embodiment_kit", POST)
def embodiment_kit(request: Request):
    print("getting data")
    data = request.json()
    print("data: ", data)
    if "messages" in data:
        resp_obj = embodiment_message_handler.handle_messages(data["messages"])
        return Response(request, json.dumps(resp_obj))
    return Response(request, json.dumps({"error": "no messages in request data"}))


while True:
    server.poll()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317

#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2026 Tim Cocks for Adafruit Industries
# SPDX-License-Identifier: MIT
"""CLI for sending commands to an MCU over Adafruit IO MQTT or HTTP and waiting for an ack."""

# pylint: disable=import-outside-toplevel, too-many-locals, too-many-return-statements
import argparse
import json
import os
import sys
import time
import uuid
from os import getenv

import requests

DEFAULT_TIMEOUT = 10  # sec
TX_FEED = "embodiment.client-to-mcu"
RX_FEED = "embodiment.mcu-to-client"
HTTP_ENDPOINT = "/embodiment_kit"


def parse_args():
    parser = argparse.ArgumentParser(
        description="Send a command to the embodiment MCU and wait for ack. Output response JSON.",
    )
    parser.add_argument(
        "command",
        nargs="?",
        help="Name of the command to send (e.g. show_squinch_face).",
    )
    parser.add_argument(
        "arguments",
        nargs="*",
        metavar="key=value",
        help="Optional arguments as key=value pairs (e.g. frequency=440 duration=0.5).",
    )
    parser.add_argument(
        "-t",
        "--timeout",
        type=float,
        default=None,
        help=f"Seconds to wait for ack (default: {DEFAULT_TIMEOUT}, or 22 for show_prompt).",
    )
    parser.add_argument(
        "-q",
        "--quiet",
        action="store_true",
        help="Suppress connection/status logging; only print the ack response.",
    )
    parser.add_argument(
        "--commandlist",
        action="store_true",
        help="Print available commands and their arguments, then exit.",
    )
    return parser.parse_args()


def parse_arguments(raw_args):
    result = {}
    for item in raw_args:
        if "=" not in item:
            raise argparse.ArgumentTypeError(
                f"Argument must be key=value format, got: {item!r}"
            )
        key, _, raw_value = item.partition("=")
        if raw_value.startswith("0x") or raw_value.startswith("0X"):
            try:
                value = int(raw_value, 16)
            except ValueError:
                value = raw_value
        else:
            try:
                value = json.loads(raw_value)
            except json.JSONDecodeError:
                value = raw_value.encode("raw_unicode_escape").decode("unicode_escape")
        result[key] = value
    return result


def build_payload(command_name, arguments, command_uuid):
    return {
        "messages": [
            {
                "metadata": {"type": "command"},
                "command": {
                    "name": command_name,
                    "arguments": arguments,
                    "uuid": command_uuid,
                },
            }
        ]
    }


def normalize_host_url(host):
    """Ensure the host has a scheme; default to http:// if missing."""
    host = host.strip().rstrip("/")
    if not host.startswith(("http://", "https://")):
        host = "http://" + host
    return host


def send_via_http(host, payload, command_uuid, timeout, log):
    """Send the command via HTTP POST to the embodiment kit device.

    Returns the ack message dict on success, or None on failure/timeout.
    """
    url = normalize_host_url(host) + HTTP_ENDPOINT
    log(f"POSTing to {url}...")

    try:
        response = requests.post(
            url,
            json=payload,
            timeout=timeout,
            headers={"Content-Type": "application/json"},
        )
    except requests.exceptions.Timeout:
        log(f"HTTP request timed out after {timeout}s")
        return None
    except requests.exceptions.RequestException as e:
        log(f"HTTP request failed: {e}")
        return None

    if response.status_code != 200:
        log(f"HTTP request returned status {response.status_code}: {response.text}")
        return None

    try:
        data = response.json()
    except ValueError as e:
        log(f"Could not decode response JSON: {e}")
        return None

    # The device responds with the same envelope format. Find the ack message
    # matching our command uuid.
    for msg in data.get("messages", []):
        metadata = msg.get("metadata", {})
        command = msg.get("command", {})
        if metadata.get("type") == "ack" and command.get("uuid") == command_uuid:
            return msg

    # If no ack-typed message is found but there's exactly one message, accept it
    # (some devices may not tag the response as ack).
    messages = data.get("messages", [])
    if len(messages) == 1:
        return messages[0]

    log("No matching ack found in HTTP response")
    return None


def send_via_aio(aio_username, aio_key, payload, command_uuid, timeout, log):
    """Send the command via Adafruit IO MQTT and wait for an ack.

    Returns the ack message dict on success, or None on timeout.
    """
    # Imported lazily so HTTP-only users don't need these packages installed.
    import adafruit_connection_manager
    import adafruit_minimqtt.adafruit_minimqtt as MQTT
    from adafruit_io.adafruit_io import IO_HTTP, IO_MQTT

    state = {"status": "waiting", "uuid": command_uuid, "response": None}

    radio = adafruit_connection_manager.CPythonNetwork()
    socket_pool = adafruit_connection_manager.get_radio_socketpool(radio)
    ssl_context = adafruit_connection_manager.get_radio_ssl_context(radio)

    # pylint: disable=unused-argument
    # The arguments aren't used, but omitting them causes exception from MQTT library
    def on_connect(client):
        client.subscribe(RX_FEED)

    def on_disconnect(client):
        pass

    def on_subscribe(client, userdata, topic, granted_qos):
        pass

    def on_unsubscribe(client, userdata, topic, pid):
        pass

    def on_message(client, feed_id, raw_payload):
        try:
            data = json.loads(raw_payload)
        except json.JSONDecodeError as e:
            log(f"Could not decode message payload: {e}")
            return

        for msg in data.get("messages", []):
            metadata = msg.get("metadata", {})
            command = msg.get("command", {})
            if metadata.get("type") == "ack" and command.get("uuid") == state["uuid"]:
                state["status"] = "received"
                state["response"] = msg
                return

    mqtt_client = MQTT.MQTT(
        broker="io.adafruit.com",
        port=8883,
        username=aio_username,
        password=aio_key,
        is_ssl=True,
        socket_pool=socket_pool,
        ssl_context=ssl_context,
    )

    io_rx = IO_MQTT(mqtt_client)
    io_rx.on_connect = on_connect
    io_rx.on_disconnect = on_disconnect
    io_rx.on_subscribe = on_subscribe
    io_rx.on_unsubscribe = on_unsubscribe
    io_rx.on_message = on_message

    io_rx.connect()

    io_tx = IO_HTTP(aio_username, aio_key, requests)
    tx_feed = io_tx.get_feed(TX_FEED)

    io_tx.send_data(tx_feed["key"], json.dumps(payload))

    start = time.monotonic()
    while time.monotonic() - start < timeout and state["status"] == "waiting":
        io_rx.loop()

    io_rx.disconnect()

    if state["status"] == "received":
        return state["response"]
    return None


def main():
    args = parse_args()

    if args.commandlist:
        help_path = os.path.join(os.path.dirname(__file__), "embodiment_cli_help.md")
        with open(help_path) as f:
            print(f.read())
        return 0

    if not args.command:
        print("ERROR: command is required.", file=sys.stderr)
        sys.exit(2)

    def log(*msg):
        if not args.quiet:
            print(*msg, file=sys.stderr)

    # Determine which transport to use. Prefer the local HTTP device if
    # EMBODIMENT_KIT_HOST is set; otherwise fall back to Adafruit IO.
    embodiment_host = getenv("EMBODIMENT_KIT_HOST")
    aio_username = getenv("ADAFRUIT_AIO_USERNAME")
    aio_key = getenv("ADAFRUIT_AIO_KEY")

    use_http = bool(embodiment_host)
    use_aio = bool(aio_username and aio_key)

    if not use_http and not use_aio:
        print(
            "ERROR: no transport configured. Set EMBODIMENT_KIT_HOST for direct HTTP, "
            "or set ADAFRUIT_AIO_USERNAME and ADAFRUIT_AIO_KEY for Adafruit IO.",
            file=sys.stderr,
        )
        sys.exit(2)

    command_uuid = str(uuid.uuid4())
    try:
        arguments = parse_arguments(args.arguments)
    except argparse.ArgumentTypeError as e:
        print(f"ERROR: {e}", file=sys.stderr)
        sys.exit(2)

    user_specified_timeout = args.timeout is not None
    if args.command == "show_prompt":
        timeout = args.timeout if user_specified_timeout else 22.0
        duration = arguments.get("timeout")
        if duration is not None:
            if user_specified_timeout and timeout <= duration:
                print(
                    f"ERROR: --timeout ({timeout}s) must be greater than"
                    + f" duration ({duration}s) for show_prompt.",
                    file=sys.stderr,
                )
                sys.exit(2)
            elif not user_specified_timeout and duration >= timeout:
                timeout = duration + 2
        args.timeout = timeout
    else:
        if not user_specified_timeout:
            args.timeout = DEFAULT_TIMEOUT

    payload = build_payload(args.command, arguments, command_uuid)

    if use_http:
        response = send_via_http(
            embodiment_host, payload, command_uuid, args.timeout, log
        )
    else:
        response = send_via_aio(
            aio_username, aio_key, payload, command_uuid, args.timeout, log
        )

    if response is not None:
        print(json.dumps(response))
        return 0

    print(
        f"ERROR: timed out after {args.timeout}s waiting for ack to uuid {command_uuid}",
        file=sys.stderr,
    )
    return 1


if __name__ == "__main__":
    sys.exit(main())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237

---
name: embodiment-kit
description: Interact with a physical embodiment kit microcontroller located in the user's room. Send commands to control a display (faces, messages, prompts), NeoPixel lights, a piezo buzzer, and a vibration motor — and to read sensors (temperature, humidity, pressure, ambient lux, microphone, accelerometer). Use to answer questions about the physical environment and to express status, get the human's attention, or communicate visually/audibly/haptically in the room.
metadata:
  version: "1.1"
  requires: ["python3", "either (ADAFRUIT_AIO_USERNAME + ADAFRUIT_AIO_KEY env vars) or (EMBODIMENT_KIT_HOST env var)"]
---

# Embodiment Kit Skill

This skill lets you read sensors from and drive actuators on a physical embodiment kit microcontroller located in the user's room. Every command round-trips and returns a JSON ack containing the current sensor readings and output state — and, for actuator commands, a `proof` block with before/after sensor readings that confirm the action had a physical effect.

## When to Use This Skill

Use this skill when you need to:

- **Sense the room**: answer questions about the current temperature, humidity, pressure, ambient light, sound level, or device orientation.
- **Express into the room**: signal status, completion, attention, or mood using the display, lights, buzzer, or vibration motor.
- **Interact with the human**: display a message they should read, or pose a multiple-choice question and wait for a button press.
- **Verify physical effects**: confirm an actuator actually fired by reading the `proof` block in the ack.

Choose actuators thoughtfully — these produce real light, sound, and motion in the human's room. Avoid firing them gratuitously or when the human may be sleeping/focused unless they've asked for it.

## Requirements

- Python virtual environment with `requests` installed.
- If using the Adafruit IO transport (see below), also need installed:
  - `adafruit-circuitpython-adafruitio`
  - `adafruit-circuitpython-connectionmanager`
  - `adafruit-circuitpython-minimqtt`
- One of the two transports below must be configured via environment variables.
- CLI script at `<skills_directory>/embodiment-kit/scripts/embodiment_cli.py`.

### Transports

The CLI supports two ways of reaching the device. Set one (the CLI prefers HTTP when both are present):

- **Direct HTTP (recommended for same-network use)**: set `EMBODIMENT_KIT_HOST` to the device's URL or hostname, e.g. `http://192.168.1.189:5000` or `embodiment.local:5000`. A scheme is optional; `http://` is assumed if missing. Port `5000` is default. The CLI POSTs the command JSON to `<host>/embodiment_kit` and reads the ack from the HTTP response — no MQTT broker or Adafruit account needed. This is the lowest-latency option and works fully offline from the public internet.
- **Adafruit IO MQTT (for remote / cross-network use)**: set both `ADAFRUIT_AIO_USERNAME` and `ADAFRUIT_AIO_KEY`. The CLI publishes the command on the `embodiment.client-to-mcu` feed and waits for the ack on `embodiment.mcu-to-client`. Use this when you cannot reach the device directly on the local network.

If neither transport's env vars are set, the CLI exits with an error before sending anything.

## How to Run a Command

General form:

```bash
python <skills_directory>/embodiment-kit/scripts/embodiment_cli.py <command> [key=value ...] [-t TIMEOUT] [-q]
```

- `key=value` arguments are parsed as JSON when possible. Integers like `0xff8800` are parsed as hex. Strings with spaces should be quoted at the shell level (`message="Hello world"`).
- `-q` / `--quiet` suppresses connection logging on stderr, leaving stdout as pure JSON suitable for piping into `jq` or `json.loads`.
- Default timeout is 10s (22s for `show_prompt`). Override with `-t SECONDS`.
- `--commandlist` prints the full command reference from `embodiment_cli_help.md`.

Every successful response is a single line of JSON on stdout, parseable directly:

```json
{
  "state": {
    "sensors": { "...": "current readings" },
    "outputs": { "neopixels": {}, "display": "..." }
  },
  "metadata": { "type": "ack", "proof": {} },
  "command": { "name": "...", "uuid": "...", "arguments": {} }
}
```

The `metadata.proof` field is only present on actuator commands. To consume the output programmatically, pipe through `jq` or read it with `json.loads`:

```bash
python embodiment_cli.py get_data -q | jq '.state.sensors.temperature'
python embodiment_cli.py get_data -q | python -c "import sys, json; print(json.load(sys.stdin)['state']['sensors']['temperature'])"
```

## Sensors (always returned in every ack)

Under `state.sensors`:

- `temperature` — degrees C
- `humidity` — percent
- `pressure` — hPa
- `lux` — ambient light
- `pdm_mic` — normalized RMS sound magnitude
- `accelerometer` — string `"x=<x>, y=<y>, z=<z> G"`, averaged over 10 samples. A device sitting flat reads roughly `z≈9.8`.

### `get_data`
Returns the standard ack with no side effects. Use to answer questions about the room's current state.

```bash
python embodiment_cli.py get_data
```

## Display Face Commands

### Preset Faces
- `show_happy_face`
- `show_sad_face`
- `show_angry_face`
- `show_confused_face`
- `show_sleepy_face`

Shows the specified preset face. Optional `background_color` (24-bit RGB int, default `0x88ddff`).

```bash
python embodiment_cli.py show_happy_face
python embodiment_cli.py show_happy_face background_color=0xffcc00
python embodiment_cli.py show_sleepy_face
python embodiment_cli.py show_sleepy_face background_color=0xffcc00
```

### `show_custom_face`
Build a face from layer choices. First call `list_custom_face_options` to see valid filenames for each layer (`eyebrows`, `eyes`, `blush`, `mouth`). Each layer is a `[filename, y_offset]` pair. The y_offset is relative to the center of the screen. Negative values move the sprite up, positive values move it down. The display height is only 135px, offset values should be within the range -50 to 50 in order to keep them visible.

```bash
python embodiment_cli.py list_custom_face_options
python embodiment_cli.py show_custom_face options='{"eyes":["round_eyes_happy.png",-20],"mouth":["mouth_smiling.png",60]}'
```

## Display Message Commands

### `show_message` — important formatting rules
Hides the face and displays text.

**Display constraints — the screen fits at most 5 rows of 19 characters per row.** Plan messages around this:

- Keep each line ≤ 19 characters. Longer lines will be cut off.
- Use at most 5 lines total. Additional lines will not render.
- Use `\n` (literal backslash-n in the shell argument) to break lines manually — do NOT rely on auto-wrapping.
- Prefer short, scannable phrasing over paragraphs. Abbreviate when needed.

```bash
python embodiment_cli.py show_message message="Build passed\nReady to deploy"
python embodiment_cli.py show_message message="Tests: 42/42 OK\nLint: clean\nCoverage: 87%"
```

Bad (too long per line, will be truncated):

```bash
# DON'T: 23 chars on line 1, no newlines — will overflow / wrap unpredictably
python embodiment_cli.py show_message message="The deployment finished successfully a moment ago"
```

### `show_prompt` — important formatting rules
Displays a message and waits for the human to press one of three hardware buttons (`D0`, `D1`, `D2`) or time out.

**Same display constraints as `show_message`: 5 rows × 19 chars.** Use them to your advantage by structuring the prompt as a question followed by labeled options on their own lines.

Recommended structure: question on line 1, then up to three options on the remaining lines, each labeled with the button index that selects it (`0)`, `1)`, `2)`):

```bash
python embodiment_cli.py show_prompt message="Deploy now?\n0) Yes\n1) No\n2) Later" timeout=30
python embodiment_cli.py show_prompt message="Mood?\n0) Good\n1) OK\n2) Tired"
```

Arguments:
- `message` (string, default `"Press a button"`).
- `timeout` (seconds, default `20`). If you set this, the CLI's own ack-timeout (`-t`) is auto-bumped to `timeout + 2`. If you override both, `-t` must be strictly greater than `timeout`.

The ack's `command.response.prompt_result` will be one of:
- `"btn D0 pressed"` → option 0 chosen
- `"btn D1 pressed"` → option 1 chosen
- `"btn D2 pressed"` → option 2 chosen
- `"timed out"` → no press before timeout

Map the button → option by the labels you wrote in the message. After a press the display briefly shows `"Thank you"`; on timeout it shows `"Prompt timed out"`.

Tips:
- Only three buttons exist — never offer more than three options.
- Keep option labels very short (e.g. `Yes`, `No`, `Later`) so the `N) label` line fits in 19 chars.
- If asking a yes/no question, you can use 2 options and leave the third button unused.
- Pick a `timeout` long enough that the human can read and decide, but not so long that the prompt blocks the conversation indefinitely (20–60s is usually right).

## Light Commands (NeoPixel strip)

### `lights_on`
Turns on the strip. `color` is a 24-bit RGB int (default magenta `0xff00ff`). `brightness` is 0.0–1.0 (default is the current brightness).

```bash
python embodiment_cli.py lights_on color=0x00ff00 brightness=0.4
```

Proof block: `idle_lux_level` vs `on_lux_level` (lux should rise).

### `lights_off`
Fills the strip with black.

```bash
python embodiment_cli.py lights_off
```

Proof block: `idle_lux_level` vs `off_lux_level` (lux should drop).

## Sound Command

### `play_tone`
Plays a tone on the piezo buzzer.

- `frequency` (Hz, default `440`)
- `duty_cycle` (default `2**15` ≈ 50%)
- `duration` (seconds, default `0.5`, **max `3`**)

```bash
python embodiment_cli.py play_tone frequency=523 duration=0.8
```

Proof block: `idle_sound_level` vs `playing_sound_level` (mic RMS should rise).

## Haptic Command

### `vibrate`
Runs the vibration motor for ~0.75 s. No arguments.

```bash
python embodiment_cli.py vibrate
```

Proof block: `idle_z_acceleration` vs `vibrating_z_acceleration` (Z-axis acceleration usually decreases during vibration).

## Patterns for Expressing Status

Some useful patterns for using actuators meaningfully:

- **Task complete, low-key**: `show_message message="Done"` or a brief `play_tone frequency=880 duration=0.2`.
- **Task complete, celebratory**: `show_happy_face` + `lights_on color=0x00ff00 brightness=0.5`, then `lights_off` a few seconds later.
- **Need attention**: `vibrate` + `show_message` describing what you need.
- **Error / warning**: `lights_on color=0xff0000 brightness=0.7` + `show_message message="Build failed\nsee terminal"`.
- **Ask the human a question**: `show_prompt` with a short question and 2–3 labeled options.
- **Ambient status indicator**: `lights_on` with a color encoding state (e.g. blue=working, green=done, red=blocked), turn off when state ends.

Always pair attention-grabbing actuators (sound, vibration, bright lights) with a display message that explains *why* you grabbed attention, so the human knows what to do next. Clean up after yourself — turn lights off when the signal is no longer relevant.

## Reference Files

- `embodiment_cli.py` — the CLI itself.
- `embodiment_cli_help.md` — the canonical command reference (also printable via `--commandlist`).
- `embodiment_message_handler.py`, `code_mcu_mqtt.py`, `code_mcu_httpserver.py` — MCU-side firmware (for context, not invoked from this skill).