🧠 Language model as the brain

{
  "acao": "girar",
  "parametros": { "graus": -90 },
  "motivo": "the user asked to turn left"
}

ollama pull <modelo>          # replace with the chosen name
ollama list                     # the model should appear here
curl http://localhost:11434/api/chat -d '{"model": "<modelo>", "stream": false,
  "messages": [{"role": "user", "content": "Reply only: ok"}]}'
python3 -m venv ~/robo-ia && source ~/robo-ia/bin/activate
pip install pyserial

#!/usr/bin/env python3
"""Local-model brain: the model CHOOSES an action from the closed list,
the code VALIDATES it, and only then sends a line of text to the ESP32 over serial.
Usage: python3 cerebro_llm.py            (with ESP32)
       python3 cerebro_llm.py --simular  (without hardware: only prints)"""
import json, sys, time, urllib.request, urllib.error

OLLAMA_URL = "http://localhost:11434/api/chat"
MODELO = "<modelo>"          # replace with the name shown by: ollama list
PORTA = "/dev/ttyUSB0"       # USB: /dev/ttyUSB0 or /dev/ttyACM0 · Pi UART: /dev/serial0 · Windows: COM3
BAUD = 115200
CONFIRMAR_ACIMA_CM = 50        # moving farther than this requires typing "s" on the keyboard

# Closed list: action -> {parameter: (minimum, maximum)} or {parameter: [permitted values]}
ACOES = {
    "andar": {"cm": (5, 100)},
    "girar": {"graus": (-180, 180)},
    "garra": {"estado": ["abrir", "fechar"]},
    "parar": {},
}

ESQUEMA = {
    "type": "object",
    "properties": {
        "acao": {"type": "string", "enum": list(ACOES)},
        "parametros": {"type": "object"},
        "motivo": {"type": "string"},
    },
    "required": ["acao", "parametros"],
}

SISTEMA = (
    "You control a small robot. Choose ONE action and respond only in JSON.\n"
    "Actions: andar {cm: 5 to 100, negative values aren’t allowed}; girar {graus: -180 to 180, "
    "negative = left}; garra {estado: open or close}; parar {}.\n"
    "If the request doesn't fit any action, or is dangerous, use stop and explain why."
)

def perguntar_modelo(pedido):
    """Send the request to Ollama and return the response dictionary (or raise an error)."""
    corpo = {
        "model": MODELO,
        "messages": [{"role": "system", "content": SISTEMA},
                     {"role": "user", "content": pedido}],
        "stream": False,           # one complete response, not pieces
        "format": ESQUEMA,          # force the JSON format
        "options": {"temperature": 0},
    }
    req = urllib.request.Request(OLLAMA_URL, data=json.dumps(corpo).encode(),
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=120) as r:
        dados = json.loads(r.read())
    return json.loads(dados["message"]["content"])

def validar(resp):
    """Return (ok, line_or_reason, needs_confirmation). Never trust the model."""
    if not isinstance(resp, dict):
        return False, "response is not a JSON object", False
    acao = resp.get("acao")
    if acao not in ACOES:
        return False, f"action '{acao}' is not on the list", False
    params = resp.get("parametros") or {}
    if not isinstance(params, dict):
        return False, "parameters are not an object", False
    sobra = set(params) - set(ACOES[acao])
    if sobra:
        return False, f"unknown parameter: {', '.join(sorted(sobra))}", False
    valores = []
    for nome, regra in ACOES[acao].items():
        if nome not in params:
            return False, f"missing parameter '{nome}'", False
        v = params[nome]
        if isinstance(regra, list):                      # allowed values
            if v not in regra:
                return False, f"'{v}' is not allowed for {nome}", False
            valores.append(str(v).upper())
        else:                                            # numeric range
            if isinstance(v, bool) or not isinstance(v, (int, float)):
                return False, f"{nome} must be a number", False
            minimo, maximo = regra
            if not (minimo <= v <= maximo):
                return False, f"{nome}={v} is outside the limit ({minimo} to {maximo})", False
            valores.append(str(int(round(v))))
    linha = " ".join([acao.upper()] + valores)         # e.g.: ANDAR 30 · GARRA ABRIR · PARAR
    confirmar = acao == "andar" and params["cm"] > CONFIRMAR_ACIMA_CM
    return True, linha, confirmar

def abrir_serial(simular):
    if simular:
        return None
    import serial                                       # pip install pyserial
    s = serial.Serial(PORTA, BAUD, timeout=2)
    time.sleep(2)                                       # opening the port resets the ESP32: wait
    s.reset_input_buffer()
    return s

def enviar(s, linha):
    if s is None:
        return f"(simulated) {linha}"
    s.write((linha + "\n").encode())
    return s.readline().decode(errors="replace").strip() or "(no response from ESP32)"

def main():
    simular = "--simular" in sys.argv
    s = abrir_serial(simular)
    print("Brain ready. Enter a request (or 'sair').")
    while True:
        pedido = input("You: ").strip()
        if pedido.lower() in ("sair", "exit"):
            break
        if not pedido:
            continue
        try:
            resp = perguntar_modelo(pedido)
        except urllib.error.URLError as e:
            print(f"Ollama did not respond ({e}). Is it running? Try: ollama list"); continue
        except (json.JSONDecodeError, KeyError) as e:
            print(f"Rejected: unreadable response ({e})."); continue
        print("Modelo:", json.dumps(resp, ensure_ascii=False))
        ok, linha, confirmar = validar(resp)
        if not ok:
            print("Recusado:", linha); continue
        if confirmar and input(f"Confirm '{linha}'? (y/n) ").strip().lower() != "s":
            print("Canceled by you."); continue
        print("ESP32:", enviar(s, linha))
    if s:
        s.write(b"PARAR\n"); s.close()

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Take a photo and ask the local VLM what's in it. pip install opencv-python"""
import base64, json, sys, urllib.request
import cv2

MODELO_VISAO = "<modelo-de-visao>"   # a model with 'vision' at ollama.com/library
PERGUNTA = sys.argv[1] if len(sys.argv) > 1 else "List the objects on the table, with their color and position (left, center, right)."

cam = cv2.VideoCapture(0)                # 0 = first camera
for _ in range(5):                       # discard dark frames at the start
    ok, quadro = cam.read()
cam.release()
if not ok:
    sys.exit("Camera didn't open: check the index (0, 1...) and the cable.")
quadro = cv2.resize(quadro, (640, 480))   # smaller image = faster response
cv2.imwrite("foto.jpg", quadro)
_, jpg = cv2.imencode(".jpg", quadro)

corpo = {
    "model": MODELO_VISAO, "stream": False, "options": {"temperature": 0},
    "messages": [{"role": "user", "content": PERGUNTA + " Answer in English, in up to 3 sentences.",
                  "images": [base64.b64encode(jpg.tobytes()).decode()]}],
}
req = urllib.request.Request("http://localhost:11434/api/chat", data=json.dumps(corpo).encode(),
                             headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=180) as r:
    print(json.loads(r.read())["message"]["content"])

// Safe executor: receives "ANDAR 30", "GIRAR -90", "GARRA ABRIR", "PARAR" (ESP32, pacote esp32 3.x)
const int PINO_PARADA = 13;              // button connected to GND (uses internal pull-up)
const unsigned long WATCHDOG_MS = 2000;  // no command for 2 s: stop
unsigned long ultimoComando = 0;
bool movendo = false;

void pararMotores() { movendo = false; Serial.println("# motors stopped"); }  // replace with your driver
void andar(int cm)    { movendo = true; Serial.printf("# moving %d cm\n", cm); }
void girar(int graus) { movendo = true; Serial.printf("# turning %d degrees\n", graus); }
void garra(bool abrir) { Serial.println(abrir ? "# opening gripper" : "# closing gripper"); }

void setup() {
  Serial.begin(115200);
  pinMode(PINO_PARADA, INPUT_PULLUP);
  pararMotores();
  Serial.println("PRONTO");
}

void loop() {
  if (digitalRead(PINO_PARADA) == LOW) { pararMotores(); while (digitalRead(PINO_PARADA) == LOW) delay(10); }
  if (movendo && millis() - ultimoComando > WATCHDOG_MS) pararMotores();
  if (!Serial.available()) return;
  String linha = Serial.readStringUntil('\n'); linha.trim();
  if (linha.length() == 0 || linha.length() > 32) { Serial.println("ERRO linha invalida"); return; }
  ultimoComando = millis();
  int esp = linha.indexOf(' ');
  String cmd = esp < 0 ? linha : linha.substring(0, esp);
  String arg = esp < 0 ? "" : linha.substring(esp + 1);
  if (cmd == "PARAR") { pararMotores(); Serial.println("OK PARAR"); }
  else if (cmd == "ANDAR") { int v = arg.toInt(); if (v < 5 || v > 100) Serial.println("ERRO limite"); else { andar(v); Serial.println("OK ANDAR"); } }
  else if (cmd == "GIRAR") { int v = arg.toInt(); if (v < -180 || v > 180) Serial.println("ERRO limite"); else { girar(v); Serial.println("OK GIRAR"); } }
  else if (cmd == "GARRA" && (arg == "ABRIR" || arg == "FECHAR")) { garra(arg == "ABRIR"); Serial.println("OK GARRA"); }
  else { pararMotores(); Serial.println("ERRO comando desconhecido"); }
}