Raspberry Pi Natural-Language GPIO Control with OpenAI: Turn On LED and Fan Using Plain English

ByJithin

Introduction

In this project, we’ll turn a Raspberry Pi into a natural-language home-automation controller. Type a plain English command in a web page and your Pi will parse it using OpenAI and trigger GPIO actions such as pulsing an LED or scheduling a fan to turn on at a specific time in Madrid.

You’ll get:

  • A minimal front-end web page with a text box.
  • A Flask backend that sends your text to OpenAI and receives a strict JSON plan.
  • Safe GPIO helpers, plus scheduling with APScheduler using the Europe/Madrid timezone.
  • A tiny script to send a simple chat request to OpenAI for your own experiments.

This follows the same practical, step-by-step style as our previous AI and Raspberry Pi tutorials so you can copy, paste, and run.

What You’ll Build

  • A local web page on your Raspberry Pi to accept commands like:
    • Turn on LED for 5 seconds
    • Turn on fan from 4 PM Madrid time
    • Turn off fan at 5 PM Madrid time
    • Turn off LED for the next 10 seconds
  • A Python backend that converts those sentences into a structured action plan and executes it on GPIO.
  • Optional offline fallback mode for quick testing without calling the API.

Components and Tools

  • Raspberry Pi with GPIO (Zero 2 W, 3B/4B, etc.)
  • LED with current-limiting resistor or a Relay Module for a fan
  • Jumper wires, breadboard or relay board
  • Internet connection for the Pi
  • OpenAI API key
  • Software:
    • Python 3
    • Flask, APScheduler, python-dotenv
    • RPi.GPIO (or system GPIO library on your distro)

Pin Plan and Wiring

  • LED on BCM 23
  • Fan relay input on BCM 24

If your relay is active-LOW, we’ll show you where to flip a single setting so ON/OFF levels are correct.

Safety tip: If you are switching mains with a relay, isolate low-voltage logic from high-voltage loads and follow proper electrical safety practices.

Prerequisites and Installation

Update your Pi and install GPIO support from apt. Then install the remaining Python packages system-wide.

sudo apt update
sudo apt install -y python3-rpi.gpio
python3 -m pip install --upgrade pip setuptools wheel --break-system-packages
python3 -m pip install Flask==3.0.3 APScheduler==3.10.4 python-dotenv "openai>=1.51.0" --break-system-packages

Set your OpenAI API key as an environment variable for this session, or place it in a .env file alongside your Python script.

export OPENAI_API_KEY=sk-your-real-key

How to Get an OpenAI API Key and Set It as an Environment Variable

Follow these steps to create your key and make it available to your Raspberry Pi apps.

Create your API key

  1. Sign in to your OpenAI account.
  2. Go to the API Keys page.
  3. Create a new secret key and copy it immediately. Keep it private and never share it.

Option A: Temporary (current terminal only)

Use this for quick tests. It resets when you close the terminal or reboot.

export OPENAI_API_KEY="sk-your-real-key-here"

Verify:

echo "$OPENAI_API_KEY"

Option B: Permanent for your user (recommended)

Add it to your shell profile so it’s available in every new terminal.

nano ~/.bashrc

Add this line at the bottom:

export OPENAI_API_KEY="sk-your-real-key-here"

Reload the profile:

source ~/.bashrc

Verify:

echo "$OPENAI_API_KEY"

Option C: Project-level using a .env file

Keep secrets next to your app and load them automatically with python-dotenv.

  1. Create a .env file inside your project folder:
OPENAI_API_KEY=sk-your-real-key-here
  1. Ensure your Python app loads it near the top:
from dotenv import load_dotenv
load_dotenv()
  1. Run your app normally. It will read the key from .env.

Option D: systemd service (auto-start on boot)

If you run your Flask app as a service, set the key in the unit file.

Edit or create a unit file:

sudo nano /etc/systemd/system/openai-gpio.service

Example service section:

[Unit]
Description=OpenAI GPIO Controller
After=network-online.target

[Service]
User=pi
WorkingDirectory=/home/pi/projects/open_ai_home_automation
ExecStart=/usr/bin/python3 /home/pi/projects/open_ai_home_automation/open_ai_home_automation.py
Environment=OPENAI_API_KEY=sk-your-real-key-here
Restart=on-failure

[Install]
WantedBy=multi-user.target

Reload and enable:

sudo systemctl daemon-reload
sudo systemctl enable --now openai-gpio.service
sudo systemctl status openai-gpio.service

Quick test: confirm the SDK sees your key

python3 - <<'PY'
import os
print("Key visible?", bool(os.getenv("OPENAI_API_KEY")))
PY

If it prints True, you’re good to go.

Project Structure

Create a folder like this:

  • open_ai_home_automation.py
  • static
    • index.html

We’ll paste both files below in Full Working Code.

How It Works

  1. You type a command in the web page.
  2. Flask sends it to OpenAI using a strict JSON schema so the model responds with predictable fields.
  3. The backend parses that plan and triggers GPIO:
    • turn_on / turn_off for immediate actions
    • pulse for on-for-N-seconds
    • schedule_on / schedule_off for future times in Europe/Madrid
  4. APScheduler runs scheduled jobs even while the app is serving requests.

If your relay is active-LOW, set ACTIVE_HIGH to False in the script to invert logic.

Run the App

Start the server:

python3 open_ai_home_automation.py

Open the printed URL in your browser and try commands like:

  • Turn on LED for 5 seconds
  • Turn on fan from 4 PM Madrid time

You’ll see the parsed plan and the execution results right below the input box.

Troubleshooting

  • 500 error in the browser: The backend prints a readable error. Check the terminal for the full traceback.
  • Authentication error: Ensure OPENAI_API_KEY is exported or present in .env.
  • Model error: Change the model name in the script if your account doesn’t have access.
  • Scheduling feels inverted: Confirm your timezone and Active-LOW vs Active-HIGH setting.
  • Reboots clear schedules: APScheduler is in-memory by default. Add a job store later if you need persistence.

Full Working Code: Backend (Flask) + Frontend (HTML)

Paste open_ai_home_automation.py in your project root:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import json
import time
import threading
import traceback
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

from flask import Flask, request, jsonify, send_from_directory
from apscheduler.schedulers.background import BackgroundScheduler
from dotenv import load_dotenv

# Load env
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
USE_LOCAL = f"{os.getenv('LOCAL_FALLBACK', '0')}" == "1"   # LOCAL_FALLBACK=1 to bypass OpenAI

# Timezone
TZ = ZoneInfo("Europe/Madrid")

# GPIO setup
ACTIVE_HIGH = True   # set False if your relay/LED is active-LOW
PIN_LED = 23         # BCM numbers
PIN_FAN = 24

try:
    import RPi.GPIO as GPIO
    ON_LEVEL = GPIO.HIGH if ACTIVE_HIGH else GPIO.LOW
    OFF_LEVEL = GPIO.LOW if ACTIVE_HIGH else GPIO.HIGH
    GPIO.setmode(GPIO.BCM)
    GPIO.setwarnings(False)
    GPIO.setup(PIN_LED, GPIO.OUT, initial=OFF_LEVEL)
    GPIO.setup(PIN_FAN, GPIO.OUT, initial=OFF_LEVEL)
    IS_PI = True
except Exception:
    class DummyGPIO:
        HIGH = 1; LOW = 0; OUT = "OUT"; BCM = "BCM"
        def setmode(self, *_): pass
        def setwarnings(self, *_): pass
        def setup(self, *_ , **__): pass
        def output(self, pin, val): print(f"[GPIO] pin {pin} -> {val}")
    GPIO = DummyGPIO()
    ON_LEVEL, OFF_LEVEL = 1, 0
    IS_PI = False

# Devices & actions
DEVICES = {"led": {"pin": PIN_LED, "type": "digital"},
           "fan": {"pin": PIN_FAN, "type": "relay"}}
VALID_ACTIONS = {"turn_on", "turn_off", "pulse", "schedule_on", "schedule_off"}

# Flask + scheduler
app = Flask(__name__, static_url_path="", static_folder="static")
sched = BackgroundScheduler(timezone=str(TZ))
sched.start()

# GPIO helpers
def gpio_set(pin: int, is_on: bool):
    GPIO.output(pin, ON_LEVEL if is_on else OFF_LEVEL)

def pulse(pin: int, seconds: float):
    gpio_set(pin, True)
    time.sleep(max(0.0, seconds))
    gpio_set(pin, False)

def schedule_job(run_at: datetime, func, job_id: str, **kwargs):
    try:
        sched.remove_job(job_id)
    except Exception:
        pass
    sched.add_job(func, "date", run_date=run_at, id=job_id, kwargs=kwargs)

# Model prompt and flat schema
SYSTEM_PROMPT = """You are an IoT command parser for a Raspberry Pi home controller.
Return ONLY JSON that matches the schema. Never include extra keys.

Allowed devices: led, fan
Allowed actions: turn_on, turn_off, pulse, schedule_on, schedule_off

ALWAYS include ALL fields for each command:
- device: "led" or "fan"
- action: one of the allowed actions
- duration_sec: number; use 0 if not applicable
- start_time: string; use "" (empty string) if not applicable (Europe/Madrid local)
- notes: string; use "" if not needed

Rules:
- If the user says "for N seconds", use action=pulse and duration_sec=N.
- If time is given without date, assume today in Europe/Madrid.
- If the time is already past for today, schedule it for the next day.
"""

JSON_SCHEMA = {
    "type": "object",
    "properties": {
        "commands": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "device": {"type": "string", "enum": ["led", "fan"]},
                    "action": {
                        "type": "string",
                        "enum": ["turn_on", "turn_off", "pulse", "schedule_on", "schedule_off"]
                    },
                    "duration_sec": {"type": "number", "minimum": 0},
                    "start_time": {"type": "string"},
                    "notes": {"type": "string"}
                },
                "required": ["device", "action", "duration_sec", "start_time", "notes"],
                "additionalProperties": False
            }
        }
    },
    "required": ["commands"],
    "additionalProperties": False
}

# Local parser for offline testing
def naive_local_parse(text: str) -> dict:
    t = text.lower()
    cmds = []
    if "led" in t and "5" in t and "second" in t and ("on" in t or "turn on" in t):
        cmds.append({"device":"led","action":"pulse","duration_sec":5,"start_time":"","notes":""})
    elif "fan" in t and "4" in t and "pm" in t and ("on" in t or "turn on" in t):
        cmds.append({"device":"fan","action":"schedule_on","duration_sec":0,"start_time":"16:00","notes":""})
    elif "fan" in t and "5" in t and "pm" in t and ("off" in t or "turn off" in t):
        cmds.append({"device":"fan","action":"schedule_off","duration_sec":0,"start_time":"17:00","notes":""})
    elif "led" in t and ("off" in t or "turn off" in t):
        cmds.append({"device":"led","action":"turn_off","duration_sec":0,"start_time":"","notes":""})
    return {"commands": cmds or [{"device":"led","action":"turn_off","duration_sec":0,"start_time":"","notes":""}]}

# OpenAI call
def call_openai(user_text: str) -> dict:
    if USE_LOCAL:
        return naive_local_parse(user_text)
    if not OPENAI_API_KEY:
        raise RuntimeError("OPENAI_API_KEY is not set. Export it or add it to a .env file.")

    from openai import OpenAI
    client = OpenAI(api_key=OPENAI_API_KEY)

    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0,
        response_format={
            "type": "json_schema",
            "json_schema": {"name": "gpio_schema", "schema": JSON_SCHEMA, "strict": True}
        },
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_text}
        ],
    )
    content = resp.choices[0].message.content
    return json.loads(content)

# Executor
def execute_command(cmd: dict):
    device = cmd["device"]; action = cmd["action"]
    dev = DEVICES.get(device)
    if not dev:
        return {"status":"error","message":f"Unknown device: {device}"}
    pin = dev["pin"]

    if action == "turn_on":
        gpio_set(pin, True);  return {"status":"ok","message":f"{device} turned on"}
    if action == "turn_off":
        gpio_set(pin, False); return {"status":"ok","message":f"{device} turned off"}
    if action == "pulse":
        duration = float(cmd.get("duration_sec", 0))
        threading.Thread(target=pulse, args=(pin, duration), daemon=True).start()
        return {"status":"ok","message":f"{device} pulsing for {duration:.2f}s"}

    if action in ("schedule_on","schedule_off"):
        start_str = (cmd.get("start_time") or "").strip()
        if not start_str:
            return {"status":"error","message":"start_time required for schedule_*"}

        try:
            if "T" in start_str or "-" in start_str:
                dt = datetime.fromisoformat(start_str)
                dt_local = dt if dt.tzinfo else dt.replace(tzinfo=TZ)
                if dt.tzinfo and dt.tzinfo != TZ:
                    dt_local = dt.astimezone(TZ)
            else:
                hh, mm = start_str.split(":")[0:2]
                now = datetime.now(TZ)
                dt_local = now.replace(hour=int(hh), minute=int(mm), second=0, microsecond=0)
        except Exception:
            return {"status":"error","message":f"Invalid start_time format: {start_str}"}

        now_local = datetime.now(TZ)
        if dt_local <= now_local:
            dt_local += timedelta(days=1)

        job_id = f"{device}-{action}-{int(dt_local.timestamp())}"
        if action == "schedule_on":
            schedule_job(dt_local, gpio_set, job_id, pin=pin, is_on=True)
            return {"status":"ok","message":f"{device} will turn ON at {dt_local.isoformat()}"}
        else:
            schedule_job(dt_local, gpio_set, job_id, pin=pin, is_on=False)
            return {"status":"ok","message":f"{device} will turn OFF at {dt_local.isoformat()}"}

    return {"status":"error","message":f"Unsupported action: {action}"}

# Routes
@app.route("/", methods=["GET"])
def index():
    return send_from_directory("static", "index.html")

@app.route("/api/command", methods=["POST"])
def api_command():
    try:
        data = request.get_json(force=True) or {}
        user_text = (data.get("text") or "").strip()
        if not user_text:
            return jsonify({"error":"Empty text"}), 400

        try:
            plan = call_openai(user_text)
        except Exception as e:
            print("[OpenAI ERROR]"); traceback.print_exc()
            return jsonify({"error": f"{type(e).__name__}: {e}"}), 500

        results = []
        for cmd in plan.get("commands", []):
            try: results.append(execute_command(cmd))
            except Exception as e: results.append({"status":"error","message":str(e)})

        return jsonify({"plan": plan, "results": results})
    except Exception as e:
        print("[API ERROR]"); traceback.print_exc()
        return jsonify({"error": f"{type(e).__name__}: {e}"}), 500

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

Create static/index.html:

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1" />
  <title>Natural-Language GPIO</title>
  <style>
    body { font-family: system-ui, Arial, sans-serif; margin: 2rem; }
    .card { max-width: 820px; border: 1px solid #e2e2e2; border-radius: 12px; padding: 1rem 1.25rem; }
    textarea { width: 100%; min-height: 120px; font-size: 16px; }
    button { padding: 0.6rem 1rem; border-radius: 10px; border: 1px solid #ccc; cursor: pointer; }
    pre { background: #f7f7f7; padding: 0.75rem; border-radius: 8px; overflow:auto; }
    ul { margin-top: 0; }
    .row { margin-top: 1rem; }
  </style>
</head>
<body>
  <h1>Natural-Language GPIO</h1>
  <div class="card">
    <p>Try:</p>
    <ul>
      <li>Turn on LED for 5 seconds</li>
      <li>Turn on fan from 4 PM Madrid time</li>
      <li>Turn off fan at 5 PM Madrid time</li>
      <li>Turn off LED for the next 10 seconds</li>
    </ul>

    <textarea id="txt" placeholder="Type your command...">Turn on LED for 5 seconds</textarea>
    <div class="row">
      <button id="send">Send</button>
    </div>

    <div id="out" class="row"></div>
  </div>

  <script>
    document.getElementById('send').onclick = async () => {
      const text = document.getElementById('txt').value.trim();
      if (!text) return;
      const res = await fetch('/api/command', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({ text })
      });
      const data = await res.json();
      if (!res.ok) {
        document.getElementById('out').innerHTML =
          `<h3>Error</h3><pre>${(data && data.error) ? data.error : 'Unknown error'}</pre>`;
        return;
      }
      document.getElementById('out').innerHTML =
        `<h3>Plan</h3><pre>${JSON.stringify(data.plan, null, 2)}</pre>
         <h3>Results</h3><pre>${JSON.stringify(data.results, null, 2)}</pre>`;
    };
  </script>
</body>
</html>

Simple Code: One-File Chat Request/Response

This is a compact script to send a single prompt to ChatGPT and print the reply.

Save as open_ai_communicate.py:

#!/usr/bin/env python3
from openai import OpenAI
import sys

client = OpenAI()  # uses OPENAI_API_KEY from your environment

prompt = " ".join(sys.argv[1:]) if len(sys.argv) > 1 else input("You: ").strip()
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": prompt}],
)
print(resp.choices[0].message.content)

Run it like:

  • python3 open_ai_communicate.py Say hi to the Raspberry Pi
  • or run with no arguments and type your message when prompted.

Final Notes and Next Steps

  • Add a simple schedule viewer and cancel buttons so you can manage upcoming jobs from the UI.
  • Extend the schema to support ranges like Turn on fan from 16:00 to 17:00 in one sentence.
  • Add device aliases such as light or lamp mapping to led.

If you want this packaged as a systemd service (auto-start on boot) with proper logs, I can add the unit file and steps next.

Similar Posts

Leave a Reply

Your email address will not be published. Required fields are marked *