Cron Agent — Scheduled Commands

A Python script that polls sensor values on a schedule and automatically sends commands when thresholds are crossed. No LLM required — pure rule-based automation via the jettyd REST API.

What you'll build

A polling agent that:

1. Reads soil moisture every 5 minutes
2. Opens the irrigation valve when moisture falls below 30%
3. Closes it when moisture exceeds 70%
4. Logs all actions to stdout (easy to pipe to a file or monitoring system)

The Python script

#!/usr/bin/env python3
"""
jettyd cron agent — moisture-triggered irrigation control.

Run with:  python3 irrigate.py
Schedule:  crontab -e → */5 * * * * /usr/bin/python3 /home/pi/irrigate.py
"""

import os
import sys
import time
import logging
import requests

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
)
log = logging.getLogger(__name__)

# ── Config ────────────────────────────────────────────────────────────────────
API_BASE   = "https://api.jettyd.com/v1"
API_KEY    = os.environ["JETTYD_API_KEY"]          # export JETTYD_API_KEY=tk_...
DEVICE_ID  = os.environ["JETTYD_DEVICE_ID"]        # export JETTYD_DEVICE_ID=uuid
METRIC     = "soil.moisture"
LOW_THRESH = 30   # open valve below this %
HIGH_THRESH = 70  # close valve above this %

HEADERS = {"Authorization": f"Bearer {API_KEY}"}

# ── Helpers ───────────────────────────────────────────────────────────────────

def read_sensor(metric: str) -> float | None:
    """Return the latest sensor value, or None on error."""
    try:
        r = requests.get(
            f"{API_BASE}/devices/{DEVICE_ID}/telemetry",
            headers=HEADERS,
            params={"metric": metric, "limit": 1},
            timeout=10,
        )
        r.raise_for_status()
        data = r.json()
        if data:
            return float(data[0]["value"])
    except Exception as exc:
        log.error("read_sensor failed: %s", exc)
    return None


def send_command(action: str, params: dict | None = None) -> bool:
    """Send a command to the device. Returns True on success."""
    try:
        r = requests.post(
            f"{API_BASE}/devices/{DEVICE_ID}/commands",
            headers=HEADERS,
            json={"command_type": action, "payload": params or {}},
            timeout=10,
        )
        r.raise_for_status()
        log.info("command sent: %s %s", action, params or "")
        return True
    except Exception as exc:
        log.error("send_command %s failed: %s", action, exc)
        return False

# ── Main ──────────────────────────────────────────────────────────────────────

def main():
    moisture = read_sensor(METRIC)
    if moisture is None:
        log.error("Could not read sensor — skipping this cycle")
        sys.exit(1)

    log.info("soil.moisture = %.1f%%", moisture)

    if moisture < LOW_THRESH:
        log.info("Below threshold (%.0f%%) — opening valve", LOW_THRESH)
        send_command("valve.on", {"duration_s": 120})
    elif moisture > HIGH_THRESH:
        log.info("Above threshold (%.0f%%) — closing valve", HIGH_THRESH)
        send_command("valve.off")
    else:
        log.info("Moisture OK — no action")

if __name__ == "__main__":
    main()

Running it

One-shot

export JETTYD_API_KEY=tk_your_key_here
export JETTYD_DEVICE_ID=your-device-uuid
python3 irrigate.py

Every 5 minutes via cron

# Edit crontab
crontab -e

# Add this line (adjust paths to match your setup):
*/5 * * * * JETTYD_API_KEY=tk_... JETTYD_DEVICE_ID=... python3 /home/pi/irrigate.py >> /var/log/irrigate.log 2>&1

As a systemd service (Raspberry Pi)

[Unit]
Description=jettyd irrigation agent
After=network.target

[Service]
Type=oneshot
ExecStart=/usr/bin/python3 /home/pi/irrigate.py
Environment=JETTYD_API_KEY=tk_your_key_here
Environment=JETTYD_DEVICE_ID=your-device-uuid
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

Enable the service with systemd timer for periodic execution.

curl alternative

No Python? Use curl directly in a shell script:

#!/usr/bin/env bash
set -euo pipefail

API_BASE="https://api.jettyd.com/v1"
HEADERS=(-H "Authorization: Bearer ${JETTYD_API_KEY}")

# Read latest moisture
MOISTURE=$(curl -sf "${HEADERS[@]}" \
  "${API_BASE}/devices/${JETTYD_DEVICE_ID}/telemetry?metric=soil.moisture&limit=1" \
  | python3 -c "import sys,json; d=json.load(sys.stdin); print(d[0]['value'] if d else 'null')")

echo "soil.moisture = ${MOISTURE}%"

if (( $(echo "$MOISTURE < 30" | bc -l) )); then
  curl -sf -X POST "${HEADERS[@]}" -H "Content-Type: application/json" \
    -d '{"command_type":"valve.on","payload":{"duration_s":120}}' \
    "${API_BASE}/devices/${JETTYD_DEVICE_ID}/commands"
  echo "Valve opened"
fi

Next: Webhook Agent Loop — Event-Driven →