• Что за зверь PicoClaw

• Установка и первый запуск

• Отображение статуса

  • Где брать статус

  • Встроенная LED-матрица

  • MPU и MCU

  • Контейнеры и сеть

  • Собираем воедино

• Заключение

"agents": {

    "defaults": {

      "workspace": "/home/arduino/.picoclaw/workspace",

      "restrict_to_workspace": true,

      "allow_read_outside_workspace": false,

      "provider": "qwen",

      "model": "qwen3-4b-thinking-2507",

      "max_tokens": 32768,

      "max_tool_iterations": 50,

      "summarize_message_threshold": 20,

      "summarize_token_percent": 75

    }

Секция model_list:

"model_list": [

    {

      "model_name": "qwen3-4b-thinking-2507",

      "model": "qwen/qwen3-4b-thinking-2507",

      "api_base": "http://192.168.88.6:1234/v1",

      "api_key": ""

    }

],

import http.server, json, re, time

LOG = "/home/arduino/picoclaw.log"

RE_IN  = re.compile(r"Processing message from")

RE_LLM = re.compile(r"LLM response|Routed message")

RE_OUT = re.compile(r"Published outbound response")

_state = "ready"

statetime = time.time()

IDLE_AFTER = 4.0

_pos = 0

def seek_end():

    global _pos

    try:

        with open(LOG) as f:

            f.seek(0, 2)

            _pos = f.tell()

    except:

        _pos = 0

def update_state():

    global state, state_time, _pos

    try:

        with open(LOG) as f:

            f.seek(_pos)

            lines = f.read().splitlines()

            _pos = f.tell()

        for line in lines:

            if RE_IN.search(line):

                state = "receiving"; state_time = time.time()

            elif RE_LLM.search(line):

                state = "thinking"; state_time = time.time()

            elif RE_OUT.search(line):

                state = "sending"; state_time = time.time()

    except:

        pass

    if state != "ready" and time.time() - state_time > IDLE_AFTER:

        _state = "ready"

class H(http.server.BaseHTTPRequestHandler):

    def do_GET(self):

        update_state()

        body = json.dumps({"state": _state}).encode()

        self.send_response(200)

        self.send_header("Content-Type", "application/json")

        self.end_headers()

        self.wfile.write(body)

    def log_message(self, *a):

        pass

seek_end()

print("State server :18791")

http.server.HTTPServer(("0.0.0.0", 18791), H).serve_forever()

void packAndDisplay(const uint16_t* rows) {

  uint32_t f[4] = {0, 0, 0, 0};

  for (int row = 0; row < 8; row++) {

    for (int col = 0; col < 13; col++) {

      if (rows[row] & (1 << (12 - col))) {

        int globalBit = row * 13 + col;

        int word = globalBit / 32;

        int pos  = 31 - (globalBit % 32);

        f[word] |= (1UL << pos);

      }

    }

  }

  matrix.loadFrame(f);

}

void clearMatrix() {

  uint32_t f[4] = {0, 0, 0, 0};

  matrix.loadFrame(f);

}

int8_t  dropY[13];

int8_t  dropLen[13];

uint8_t dropSpeed[13];

uint8_t dropTick[13];

void matrixInit() {

  for (int c = 0; c < 13; c++) {

    dropY[c]     = -(random(0, 10));

    dropLen[c]   = random(2, 6);

    dropSpeed[c] = random(1, 4);

    dropTick[c]  = 0;

  }

}

void matrixStep() {

  uint16_t rows[8] = {0};

  for (int c = 0; c < 13; c++) {

    dropTick[c]++;

    if (dropTick[c] >= dropSpeed[c]) {

      dropTick[c] = 0;

      dropY[c]++;

      if (dropY[c] - dropLen[c] > 7) {

        dropY[c]     = -(random(0, 8));

        dropLen[c]   = random(2, 6);

        dropSpeed[c] = random(1, 4);

      }

    }

    for (int i = 0; i < dropLen[c]; i++) {

      int y = dropY[c] - i;

      if (y >= 0 && y < 8) {

        rows[y] |= (1 << (12 - c));

      }

    }

  }

  packAndDisplay(rows);

}

Статичная иконка краба рисуется следующим образом:

const uint16_t crab[8] = {

  0b0100000000010,

  0b0010000000100,

  0b0011011101100,

  0b0001111111000,

  0b0011101011100,

  0b0011111111100,

  0b0010101010100,

  0b0010000000100,

};

def detect_host_ip():

    override = os.environ.get("PICOCLAW_HOST")

    if override:

        print("[net] PICOCLAW_HOST override: {}".format(override), flush=True)

        return override

    try:

        with open("/proc/net/route") as f:

            for line in f.readlines()[1:]:

                fields = line.strip().split()

                # destination == 00000000 → default route

                if len(fields) >= 3 and fields[1] == "00000000":

                    gw_hex = fields[2]

                    ip = ".".join(

                        str(int(gw_hex[i:i+2], 16))

                        for i in range(6, -1, -2)

                    )

                    print("[net] default gateway detected: {}".format(ip), flush=True)

                    return ip

    except Exception as e:

        print("[net][ERROR] detect_host_ip: {}: {}".format(

            type(e).__name__, e), flush=True)

    # Запасной вариант - дефолтный docker bridge

    print("[net] fallback to 172.17.0.1", flush=True)

    return "172.17.0.1"

HOST            = detect_host_ip()

HEALTH_URL      = "http://{}:18790/health".format(HOST)

STATE_URL       = "http://{}:18791/".format(HOST)

POLL_INTERVAL   = 0.5

SENDING_TIMEOUT = 3.0

#!/bin/bash

stop_all() {

    echo "Останавливаем..."

    pkill -f "picoclaw gateway" 2>/dev/null

    pkill -f state_server.py 2>/dev/null

    pkill -f "tee /home/arduino/picoclaw.log" 2>/dev/null

    echo "Готово"

}

start_all() {

    stop_all

    sleep 1

    picoclaw gateway 2>&1 | tee /home/arduino/picoclaw.log &

    sleep 3

    python3 /home/arduino/state_server.py &

    echo "Всё запущено"

}

case "$1" in

    stop)  stop_all ;;

    *)     start_all ;;

esac

main.py

from arduino.app_utils import *

import time

import os

import traceback

try:

    import urllib.request

    import json

    HAS_HTTP = True

except ImportError:

    HAS_HTTP = False

    print("[WARNING] urllib недоступен", flush=True)

def detect_host_ip():

    override = os.environ.get("PICOCLAW_HOST")

    if override:

        print("[net] PICOCLAW_HOST override: {}".format(override), flush=True)

        return override

    try:

        with open("/proc/net/route") as f:

            for line in f.readlines()[1:]:

                fields = line.strip().split()

                if len(fields) >= 3 and fields[1] == "00000000":

                    gw_hex = fields[2]

                    ip = ".".join(

                        str(int(gw_hex[i:i+2], 16))

                        for i in range(6, -1, -2)

                    )

                    print("[net] default gateway detected: {}".format(ip), flush=True)

                    return ip

    except Exception as e:

        print("[net][ERROR] detect_host_ip: {}: {}".format(

            type(e).__name__, e), flush=True)

    print("[net] fallback to 172.17.0.1", flush=True)

    return "172.17.0.1"

HOST = detect_host_ip()

HEALTH_URL = "http://{}:18790/health".format(HOST)

STATE_URL = "http://{}:18791/".format(HOST)

POLL_INTERVAL = 0.5

SENDING_TIMEOUT = 3.0

currentanim = None

def set_led(name, force=False):

    global currentanim

    if currentanim == name and not force:

        return

    currentanim = name

    try:

        result = Bridge.call("set_animation", name)

        print("[LED] -> {} (bridge: {!r})".format(name, result), flush=True)

    except Exception as e:

        print("[LED][ERROR] set_animation({}): {}: {}".format(

            name, type(e).__name__, e), flush=True)

        traceback.print_exc()

httperr_logged = False

def http_get(url):

    global httperr_logged

    try:

        resp = urllib.request.urlopen(url, timeout=1)

        return json.loads(resp.read().decode())

    except Exception as e:

        if not httperr_logged:

            print("[HTTP][ERROR] {} -> {}: {}".format(

                url, type(e).__name__, e), flush=True)

            httperr_logged = True

        return None

def http_ok_again():

    global httperr_logged

    if httperr_logged:

        print("[HTTP] связь восстановлена", flush=True)

        httperr_logged = False

_started = False

picoclawalive = False

misscount = 0

MISS_MAX = 4

lastsending_t = 0

prevstate = ""

def loop():

    global picoclawalive, misscount, lastsending_t, prevstate, _started

    if not _started:

        _started = True

        print("[start] PicoClaw LED monitor", flush=True)

        print("[start] HEALTH_URL = {}".format(HEALTH_URL), flush=True)

        print("[start] STATE_URL  = {}".format(STATE_URL), flush=True)

        set_led("thinking", force=True)

    health = http_get(HEALTH_URL)

    if health is None:

        misscount += 1

        if misscount == MISS_MAX:

            print("[poll] picoclaw не отвечает", flush=True)

        if misscount >= MISS_MAX and picoclawalive:

            picoclawalive = False

            set_led("clear")

        time.sleep(POLL_INTERVAL)

        return

    http_ok_again()

    misscount = 0

    if not picoclawalive:

        print("[poll] picoclaw появился", flush=True)

        picoclawalive = True

    state_data = http_get(STATE_URL)

    if state_data is None:

        set_led("ready")

        time.sleep(POLL_INTERVAL)

        return

    state = state_data.get("state", "ready")

    if state == "sending":

        if prevstate != "sending":

            lastsending_t = time.time()

        set_led("sending")

    elif state == "thinking":

        set_led("thinking")

    elif state == "receiving":

        set_led("receiving")

    else:

        if currentanim == "sending":

            if time.time() - lastsending_t < SENDING_TIMEOUT:

                pass

            else:

                set_led("ready")

        else:

            set_led("ready")

    prevstate = state

    time.sleep(POLL_INTERVAL)

App.run(user_loop=loop)

sketch.ino

#include "Arduino_LED_Matrix.h"

#include "Arduino_RouterBridge.h"

ArduinoLEDMatrix matrix;

enum State {

  STATE_CLEAR,

  STATE_THINKING,

  STATE_RECEIVING,

  STATE_SENDING,

  STATE_READY

};

volatile State currentState = STATE_CLEAR;

volatile bool  stateChanged = false;

unsigned long lastFrameTime = 0;

int  frameIndex             = 0;

bool animationDone          = false;

const unsigned long MATRIX_SPEED   = 80;

const unsigned long ANTENNA_SPEED   = 200;

const unsigned long CRAB_SPEED     = 500;

void packAndDisplay(const uint16_t* rows) {

  uint32_t f[4] = {0, 0, 0, 0};

  for (int row = 0; row < 8; row++) {

    for (int col = 0; col < 13; col++) {

      if (rows[row] & (1 << (12 - col))) {

        int globalBit = row * 13 + col;

        int word = globalBit / 32;

        int pos  = 31 - (globalBit % 32);

        f[word] |= (1UL << pos);

      }

    }

  }

  matrix.loadFrame(f);

}

void clearMatrix() {

  uint32_t f[4] = {0, 0, 0, 0};

  matrix.loadFrame(f);

}

int8_t  dropY[13];      

int8_t  dropLen[13];    

uint8_t dropSpeed[13];  

uint8_t dropTick[13];    

void matrixInit() {

  for (int c = 0; c < 13; c++) {

    dropY[c]     = -(random(0, 10));  

    dropLen[c]   = random(2, 6);

    dropSpeed[c] = random(1, 4);

    dropTick[c]  = 0;

  }

}

void matrixStep() {

  uint16_t rows[8] = {0};

  for (int c = 0; c < 13; c++) {

    dropTick[c]++;

    if (dropTick[c] >= dropSpeed[c]) {

      dropTick[c] = 0;

      dropY[c]++;

     

      if (dropY[c] - dropLen[c] > 7) {

        dropY[c]     = -(random(0, 8));

        dropLen[c]   = random(2, 6);

        dropSpeed[c] = random(1, 4);

      }

    }

   

    for (int i = 0; i < dropLen[c]; i++) {

      int y = dropY[c] - i;

      if (y >= 0 && y < 8) {

        rows[y] |= (1 << (12 - c));

      }

    }

  }

  packAndDisplay(rows);

}

const uint16_t ant0[8] = {

  0b0000000000000,  

  0b0000000000000,  

  0b0000000000000,  

  0b0000000000000,  

  0b0000000000000,  

  0b0000001000000,  

  0b0000001000000,  

  0b0000011100000,  

};

const uint16_t ant1[8] = {

  0b0000000000000,  

  0b0000000000000,  

  0b0000000000000,  

  0b0000010100000,  

  0b0000001000000,  

  0b0000001000000,  

  0b0000001000000,  

  0b0000011100000,  

};

const uint16_t ant2[8] = {

  0b0000000000000,  

  0b0000100010000,  

  0b0000010100000,  

  0b0000000000000,  

  0b0000010100000,  

  0b0000001000000,  

  0b0000001000000,  

  0b0000011100000,  

};

const uint16_t ant3[8] = {

  0b0001000001000,  

  0b0000100010000,  

  0b0000000000000,  

  0b0000100010000,  

  0b0000010100000,  

  0b0000001000000,  

  0b0000001000000,  

  0b0000011100000,  

};

const uint16_t ant4[8] = {

  0b0010000000100,  

  0b0001000001000,  

  0b0000000000000,  

  0b0001000001000,  

  0b0000100010000,  

  0b0000010100000,  

  0b0000001000000,  

  0b0000011100000,  

};

const uint16_t ant5[8] = {

  0b0100000000010,  

  0b0010000000100,  

  0b0000000000000,  

  0b0010000000100,  

  0b0001000001000,  

  0b0000100010000,  

  0b0000001000000,  

  0b0000011100000,  

};

const uint16_t* ant_send_frames[] = { ant0, ant1, ant2, ant3, ant4, ant5 };

const uint16_t* ant_recv_frames[] = { ant5, ant4, ant3, ant2, ant1, ant0 };

const int ANT_COUNT = 6;

const uint16_t crab[8] = {

  0b0100000000010,  

  0b0010000000100,  

  0b0011011101100,  

  0b0001111111000,  

  0b0011101011100,  

  0b0011111111100,  

  0b0010101010100,  

  0b0010000000100,  

};

String set_animation(String anim) {

  State newState = STATE_CLEAR;

  if      (anim == "thinking")  newState = STATE_THINKING;

  else if (anim == "receiving") newState = STATE_RECEIVING;

  else if (anim == "sending")   newState = STATE_SENDING;

  else if (anim == "ready")     newState = STATE_READY;

  else if (anim == "clear")     newState = STATE_CLEAR;

  else return String("unknown");

  currentState = newState;

  stateChanged = true;

  return anim;

}

void setup() {

  matrix.begin();

  clearMatrix();

  randomSeed(analogRead(A0));

  Bridge.begin();

  Monitor.begin();

  if (!Bridge.provide("set_animation", set_animation)) {

    Monitor.println("Error: failed to provide set_animation");

  } else {

    Monitor.println("Registered: set_animation");

  }

}

void loop() {

  if (stateChanged) {

    stateChanged  = false;

    frameIndex    = 0;

    animationDone = false;

    lastFrameTime = millis();

    if (currentState == STATE_CLEAR) {

      clearMatrix();

    }

    if (currentState == STATE_THINKING) {

      matrixInit();

    }

    if (currentState == STATE_READY) {

      packAndDisplay(crab);

    }

  }

  if (currentState == STATE_CLEAR) {

    delay(10);

    return;

  }

 

  if (currentState == STATE_READY) {

    delay(50);

    return;

  }

  unsigned long speed;

  switch (currentState) {

    case STATE_THINKING:  speed = MATRIX_SPEED;   break;

    case STATE_RECEIVING: speed = ANTENNA_SPEED; break;

    case STATE_SENDING:   speed = ANTENNA_SPEED; break;

    default: return;

  }

  if (millis() - lastFrameTime < speed) return;

  lastFrameTime = millis();

  switch (currentState) {

    case STATE_THINKING:

      matrixStep();

      break;

    case STATE_RECEIVING:

      if (frameIndex >= ANT_COUNT) { animationDone = true; frameIndex = ANT_COUNT - 1; }

      packAndDisplay(ant_recv_frames[frameIndex]);

      break;

    case STATE_SENDING:

      if (frameIndex >= ANT_COUNT) { animationDone = true; frameIndex = ANT_COUNT - 1; }

      packAndDisplay(ant_send_frames[frameIndex]);

      break;

    default: break;

  }

  if (!animationDone) frameIndex++;

}

sketch.yaml

profiles:

  default:

    fqbn: 

    platforms:

      - platform: arduino:zephyr

    libraries:

      - ArxContainer (0.7.0)

      - ArxTypeTraits (0.3.2)

      - DebugLog (0.8.4)

      - MsgPack (0.4.2)

      - Arduino_RPClite (0.2.1)

      - Arduino_RouterBridge (0.4.0)

      - ArduinoGraphics (1.1.5)

default_profile: default