Quick start

pip install paho-mqtt

# Provision the PASSWORD credential for a plant. Returns a JSON body with
# the generated username (plc-<first 8 hex>) + password — shown once.
curl -X POST https://api.voke.lovinka.com/api/v1/plants/{plantId}/provision \
  -H "Content-Type: application/json" \
  -H "Cookie: <your session cookie>" \
  -d '{"method": "PASSWORD"}'

# hmac_sign.py
"""Canonical HMAC-SHA256 helpers for PLC message signing.

Mirrors packages/shared/src/utils/hmac.ts exactly. If you change one side,
change both and re-check the known-value tests.

Key facts (verified against the TypeScript source):
  - Separator:  "|"  (pipe, U+007C)
  - Telemetry:  deviceId | ts | nonce | canonical_json_stringify(data)
  - Command:    deviceId | cmdId | ts | action | canonical_json_stringify(payload)
  - Ack:        deviceId | cmdId | ts | status | nonce
  - Alarm:      deviceId | ts | nonce | ev | alarmId | code | sev

  The telemetry payload is the message body dict with ts/n/sig stripped out,
  serialised with sorted keys (recursive) and no spaces — matching
  canonicalJsonStringify() in hmac.ts.

Dependencies: Python 3.11+ standard library only.
"""
from __future__ import annotations

import hashlib
import hmac
import json
from typing import Any

# ─── Canonical JSON ───────────────────────────────────────────────────────────

def canonical_json_stringify(obj: Any) -> str:  # noqa: ANN401
    """Recursively serialise *obj* with sorted keys and no spaces."""
    if obj is None:
        return "null"
    if isinstance(obj, bool):
        return "true" if obj else "false"
    if isinstance(obj, (int, float)):
        return json.dumps(obj)
    if isinstance(obj, str):
        return json.dumps(obj)
    if isinstance(obj, list):
        return "[" + ",".join(canonical_json_stringify(item) for item in obj) + "]"
    if isinstance(obj, dict):
        sorted_keys = sorted(obj.keys())
        parts = [
            json.dumps(k) + ":" + canonical_json_stringify(obj[k])
            for k in sorted_keys
        ]
        return "{" + ",".join(parts) + "}"
    return json.dumps(obj)


# ─── Core primitives ─────────────────────────────────────────────────────────

_SEP = "|"


def compute_signature(parts: list[str], secret: str) -> str:
    message = _SEP.join(parts).encode("utf-8")
    return hmac.new(secret.encode("utf-8"), message, hashlib.sha256).hexdigest()


def verify_signature(parts: list[str], secret: str, signature: str) -> bool:
    expected = compute_signature(parts, secret)
    if len(expected) != len(signature):
        return False
    return hmac.compare_digest(expected, signature)


# ─── Domain signers ───────────────────────────────────────────────────────────

def sign_telemetry(device_id, ts, nonce, data, secret):
    """Sign: deviceId | ts | nonce | canonicalJsonStringify(data)"""
    return compute_signature(
        [device_id, str(ts), nonce, canonical_json_stringify(data)],
        secret,
    )


def sign_command_ack(device_id, cmd_id, ts, status, nonce, secret):
    """Sign: deviceId | cmdId | ts | status | nonce"""
    return compute_signature([device_id, cmd_id, str(ts), status, nonce], secret)

# quick_connect.py
import paho.mqtt.client as mqtt

# ── Configuration ──────────────────────────────────────────────────────────────
HOST = "mqtt.voke.lovinka.com"   # or "localhost" for local dev
PORT = 8885                       # PASSWORD listener — same port prod and dev
PLANT_ID = "your-plant-uuid"     # UUID from Voke admin
MQTT_PASSWORD = "your-password"  # generated in Step 2
CA_CERT = None                    # set to "/path/to/ca.crt" for local dev

# ── Client setup ──────────────────────────────────────────────────────────────
client = mqtt.Client(client_id=PLANT_ID, protocol=mqtt.MQTTv5)
client.username_pw_set(PLANT_ID, MQTT_PASSWORD)

if CA_CERT:
    # Local dev: trust the Voke CA explicitly
    client.tls_set(ca_certs=CA_CERT)
else:
    # Production: use system trust store
    client.tls_set()

client.connect(HOST, PORT, keepalive=60)

# publish_snapshot.py
import json
import secrets
import time

import paho.mqtt.client as mqtt

from hmac_sign import sign_telemetry

PLANT_ID = "REPLACE-WITH-YOUR-PLANT-ID"
TELEMETRY_TOPIC = f"cpi/{PLANT_ID}/telemetry"

# HMAC secret provisioned via Voke admin → Plant → Security.
# PASSWORD devices may omit signing if your ACL policy permits;
# mTLS and JWT devices must sign.
HMAC_SECRET = "REPLACE-WITH-YOUR-DEVICE-SECRET"


def build_snapshot_body() -> dict:
    """Build the inner payload — the part that gets HMAC-signed.

    Keys inside ``devices`` entries must match your DeviceTemplate:
      - Cabinet ``raw``:    unsigned 32-bit int; each bit maps to a binary signal.
      - Meter/inverter/battery ``values``: field names mapped to numeric signals via template.field.

    externalId conventions from seed data:
      Cabinets  → R1, R2, …   (binary signals)
      Meters    → GRID, E1, … (numeric signals)
      Inverters → INV1, PV1, M1, … (numeric signals — match whatever is enrolled)
      Batteries → BAT1, …     (numeric signals)
    """
    return {
        # ISO-8601 timestamp (optional — Voke falls back to the envelope ``ts``).
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "devices": [
            {
                # Cabinet R1: bits 0 and 2 are high → 0b00000101 = 5
                "externalId": "R1",
                "type": "CABINET",
                "raw": 5,
            },
            {
                # Meter M1: field names must match template signal.field
                "externalId": "M1",
                "type": "METER",
                "values": {
                    "activePowerKw": 12.4,
                    "voltageV": 231.7,
                },
            },
            {
                # Battery BAT1: same numeric values contract
                "externalId": "BAT1",
                "type": "BATTERY",
                "values": {
                    "stateOfChargePct": 61.2,
                    "batteryPowerW": -820,
                },
            },
        ],
    }


def publish_snapshot(client: mqtt.Client) -> None:
    ts = int(time.time() * 1000)   # Unix milliseconds
    nonce = secrets.token_hex(8)    # 16 hex chars; must be unique per message
    body = build_snapshot_body()

    sig = sign_telemetry(PLANT_ID, ts, nonce, body, HMAC_SECRET)

    # Final envelope: flat object — envelope fields + body fields merged.
    message = {"ts": ts, "n": nonce, "sig": sig, **body}
    payload = json.dumps(message, separators=(",", ":"))

    client.publish(TELEMETRY_TOPIC, payload, qos=1)
    print(f"[snapshot] published: {json.dumps(message, indent=2)}")


def on_connect(client, _userdata, _flags, rc, _props=None):
    if rc == 0:
        print("[connect] connected")
        publish_snapshot(client)
    else:
        print(f"[connect] failed: rc={rc}")


if __name__ == "__main__":
    client = mqtt.Client(client_id=PLANT_ID, protocol=mqtt.MQTTv5)
    client.username_pw_set(PLANT_ID, MQTT_PASSWORD)
    client.tls_set()   # or tls_set(ca_certs="/path/to/ca.crt") for local dev
    client.on_connect = on_connect
    client.connect("mqtt.voke.lovinka.com", 8885, keepalive=60)
    client.loop_forever()

{
  "ts": 1713540000000,
  "n": "0a1b2c3d4e5f6789",
  "sig": "abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890",
  "timestamp": "2026-04-19T14:00:00Z",
  "devices": [
    {"externalId": "R1", "type": "CABINET", "raw": 5},
    {"externalId": "M1", "type": "METER", "values": {"activePowerKw": 12.4, "voltageV": 231.7}},
    {"externalId": "BAT1", "type": "BATTERY", "values": {"stateOfChargePct": 61.2, "batteryPowerW": -820}}
  ]
}

# Query sub-device telemetry for the last hour
curl "https://api.voke.lovinka.com/api/v1/telemetry/sub-device\
?plantId=YOUR_PLANT_ID\
&subDeviceExternalId=M1\
&metric=activePowerKw\
&from=2026-04-19T13:00:00Z\
&to=2026-04-19T14:00:00Z" \
  -H "Cookie: <your session cookie>"

# subscribe_commands.py
import json
import secrets
import time
from typing import Any

import paho.mqtt.client as mqtt

from hmac_sign import sign_command_ack, verify_signature, canonical_json_stringify

PLANT_ID = "REPLACE-WITH-YOUR-PLANT-ID"
COMMAND_TOPIC = f"cpi/{PLANT_ID}/command"
ACK_TOPIC = f"cpi/{PLANT_ID}/ack"
HMAC_SECRET = "REPLACE-WITH-YOUR-DEVICE-SECRET"


def verify_command_sig(cmd: dict[str, Any]) -> bool:
    """Verify Voke's signature on an inbound PlcCommand.

    Part order: deviceId | cmdId | ts | action | canonicalJsonStringify(payload)
    """
    sig = cmd.get("sig", "")
    if not sig:
        return False
    parts = [
        PLANT_ID,
        str(cmd["cmdId"]),
        str(cmd["ts"]),
        str(cmd["type"]),
        canonical_json_stringify(cmd.get("p", {})),
    ]
    return verify_signature(parts, HMAC_SECRET, sig)


def send_ack(client, cmd_id, status, *, err=None, msg=None):
    ts = int(time.time() * 1000)
    nonce = secrets.token_hex(8)
    sig = sign_command_ack(PLANT_ID, cmd_id, ts, status, nonce, HMAC_SECRET)
    ack: dict[str, Any] = {"cmdId": cmd_id, "st": status, "ts": ts, "n": nonce, "sig": sig}
    if err:
        ack["err"] = err
    if msg:
        ack["msg"] = msg
    client.publish(ACK_TOPIC, json.dumps(ack, separators=(",", ":")), qos=1)
    print(f"[cmd] ACK sent: cmdId={cmd_id} st={status}")


def on_message(client, _userdata, msg):
    try:
        cmd = json.loads(msg.payload.decode())
    except json.JSONDecodeError:
        print(f"[cmd] malformed payload on {msg.topic}")
        return

    cmd_id = cmd.get("cmdId")
    cmd_type = cmd.get("type")
    print(f"[cmd] received: cmdId={cmd_id} type={cmd_type}")

    if not cmd_id or not cmd_type:
        return

    if HMAC_SECRET and not verify_command_sig(cmd):
        print(f"[cmd] invalid signature on cmdId={cmd_id} — discarding")
        return

    # Acknowledge receipt immediately.
    send_ack(client, cmd_id, "RECEIVED")

    # Replace with real hardware dispatch; send COMPLETED or FAILED when done.
    send_ack(client, cmd_id, "COMPLETED")


def on_connect(client, _userdata, _flags, rc, _props=None):
    if rc == 0:
        client.subscribe(COMMAND_TOPIC, qos=1)
        client.on_message = on_message
        print(f"[cmd] subscribed to {COMMAND_TOPIC}")


if __name__ == "__main__":
    MQTT_PASSWORD = "REPLACE-WITH-YOUR-MQTT-PASSWORD"
    client = mqtt.Client(client_id=PLANT_ID, protocol=mqtt.MQTTv5)
    client.username_pw_set(PLANT_ID, MQTT_PASSWORD)
    client.tls_set()
    client.on_connect = on_connect
    client.connect("mqtt.voke.lovinka.com", 8885, keepalive=60)
    client.loop_forever()