Version: 0.1.0-draft
Author: Agent B — Device Mesh & IoT Layer
Date: 2026-04-04
Status: Design Document
CloseClaw is the inverse of OpenClaw. Where OpenClaw reaches outward to the cloud, CloseClaw turns inward — a fully offline, local-only AI agent framework for smart homes, robotics, and embedded devices. It never touches the internet. Not once. Not ever.
- Device Mesh Protocol
- Home Automation Integration
- Voice Interface
- Sensor/Actuator Abstraction
- Security Model
- Edge Inference Optimization
- Reference Architecture
- Implementation Roadmap
- Zero cloud dependency — all discovery, routing, and messaging is LAN-local
- Self-healing — mesh reconfigures when nodes join or leave without manual intervention
- Encrypted by default — no plaintext traffic between nodes, ever
- Hardware-agnostic — runs on Pi Zero 2W, ESP32 (with adapter), NUC, NAS, or x86 tower
- Bandwidth-conscious — assumes heterogeneous links: Ethernet, Wi-Fi, Zigbee bridges, even serial
Every CloseClaw node announces itself on the local network using mDNS (multicast DNS, RFC 6762) with a custom service type:
_closeclaw._tcp.local
{
"service": "_closeclaw._tcp.local",
"name": "closeclaw-hub-01",
"host": "closeclaw-hub-01.local",
"port": 7700,
"txt": {
"version": "0.1.0",
"role": "hub",
"capabilities": "inference,voice,gpio,camera",
"mesh_id": "a3f7b21c",
"pubkey_fingerprint": "sha256:b4e9...",
"api": "grpc+tls"
}
}Fields:
# closeclaw/mesh/discovery.py
import zeroconf
from zeroconf import ServiceBrowser, Zeroconf
class MeshDiscovery:
SERVICE_TYPE = "_closeclaw._tcp.local."
def __init__(self, node_identity: NodeIdentity):
self.zc = Zeroconf()
self.identity = node_identity
self.peers: dict[str, PeerInfo] = {}
def start(self):
# Announce self
info = self._build_service_info()
self.zc.register_service(info)
# Discover peers
self.browser = ServiceBrowser(self.zc, self.SERVICE_TYPE, self)
def add_service(self, zc, type_, name):
info = zc.get_service_info(type_, name)
peer = self._parse_peer(info)
if peer and self._verify_mesh_id(peer):
self.peers[peer.node_id] = peer
EventBus.emit("mesh.peer_joined", peer)
def remove_service(self, zc, type_, name):
if name in self.peers:
EventBus.emit("mesh.peer_left", self.peers.pop(name))On Linux: Uses Avahi daemon as the underlying mDNS responder (avahi-daemon). CloseClaw ships a minimal avahi-daemon.conf template.
On macOS (dev): Uses the native Bonjour stack.
On embedded (no mDNS): Falls back to static peer list in config/peers.yaml + UDP broadcast ping on port 7701.
CloseClaw uses gRPC over mutual TLS (mTLS) as its primary inter-node transport. All certificates are issued by the mesh's local CA (see §5).
┌─────────────┐ gRPC/mTLS ┌─────────────┐ MQTT/TLS ┌──────────────┐
│ Robot Node │ ────────────▶│ Hub Node │ ────────────▶│ HAOS Bridge │
│ (Pi 4) │ │ (NUC/NAS) │ │ (HA VM) │
└─────────────┘ └─────────────┘ └──────────────┘
│ │
│ gRPC/mTLS │ gRPC/mTLS
▼ ▼
┌─────────────┐ ┌─────────────┐
│ Camera │ │ Voice Node │
│ (Pi Zero) │ │ (Pi 4) │
└─────────────┘ └─────────────┘
// closeclaw/proto/mesh.proto
syntax = "proto3";
package closeclaw.mesh;
service MeshNode {
// Bidirectional stream for real-time events
rpc EventStream(stream NodeEvent) returns (stream NodeEvent);
// Request/response for commands
rpc Execute(ExecuteRequest) returns (ExecuteResponse);
// Sensor data push
rpc PushSensorData(SensorReading) returns (Ack);
// Inference delegation (send to hub)
rpc Infer(InferRequest) returns (stream InferChunk);
}
message NodeEvent {
string event_id = 1;
string source_node = 2;
string event_type = 3; // "sensor.motion", "voice.utterance", "device.state"
bytes payload = 4; // MessagePack-encoded event data
int64 timestamp_ms = 5;
bytes signature = 6; // Ed25519 signature over fields 1-5
}
message InferRequest {
string model_hint = 1; // "small", "medium", "large", or model name
bytes prompt = 2;
map<string, bytes> context = 3;
bool stream = 4;
int32 max_tokens = 5;
}For resource-constrained nodes (ESP32, microcontrollers with MQTT client libraries), CloseClaw runs an embedded Mosquitto broker on the hub. Topics follow this hierarchy:
closeclaw/<mesh_id>/<node_id>/<category>/<topic>
Examples:
closeclaw/a3f7b21c/robot-01/sensors/ultrasonic
closeclaw/a3f7b21c/hub/inference/response
closeclaw/a3f7b21c/hub/commands/robot-01/move
TLS on MQTT: Client certificates required. The hub's embedded broker is configured with require_certificate true and validates against the local CA.
The Hub acts as the mesh router for inter-node messages. Edge nodes do not route traffic — they only communicate with the hub directly (star topology with hub-to-hub peering for multi-hub setups).
Edge Node → Hub (primary) → Edge Node (target)
Hub A → Hub B (peer) → Edge Node (on Hub B's network)
Hub election: In a single-hub deployment, the hub is statically configured. In multi-hub setups, hubs elect a primary hub using a simple Bully algorithm over the mesh event stream. The primary hub holds the authoritative device registry and routes cross-hub traffic.
The internal event bus (closeclaw/mesh/eventbus.py) is the nervous system of CloseClaw. All components — inference, voice, sensors, automations — communicate through typed events:
# Event categories
class EventType:
# Sensor events
SENSOR_MOTION = "sensor.motion"
SENSOR_TEMP = "sensor.temperature"
SENSOR_DOOR = "sensor.door"
SENSOR_CAMERA = "sensor.camera.frame"
# Voice events
VOICE_WAKE = "voice.wake_word"
VOICE_UTTERANCE = "voice.utterance"
VOICE_RESPONSE = "voice.response"
# Device control
DEVICE_STATE = "device.state_changed"
DEVICE_COMMAND = "device.command"
# Inference events
INFER_REQUEST = "inference.request"
INFER_RESPONSE = "inference.response"
INFER_DELEGATED = "inference.delegated_to"
# Mesh events
MESH_PEER_JOINED = "mesh.peer_joined"
MESH_PEER_LEFT = "mesh.peer_left"
MESH_SYNC = "mesh.state_sync"Events are signed with the originating node's Ed25519 key and verified before dispatch to local subscribers.
CloseClaw does not replace Home Assistant, MQTT brokers, or Zigbee coordinators. It augments them — sitting above the existing smart home stack as the AI reasoning layer that:
- Interprets natural language commands
- Writes and executes automations
- Observes sensor streams and infers intent
- Answers questions about the home state
- Learns household patterns over time (fully local, no upload)
┌──────────────────────────────────────────────────────┐
│ CloseClaw Hub │
│ ┌───────────────┐ ┌──────────────┐ ┌──────────┐ │
│ │ AI Reasoning │ │ Automation │ │ Voice │ │
│ │ (LLM local) │ │ Engine │ │ Pipeline│ │
│ └───────────────┘ └──────────────┘ └──────────┘ │
│ │ │ │ │
│ └────────────────┴────────────────┘ │
│ Event Bus │
│ │ │ │ │
│ ┌─────────▼──────┐ ┌──────▼───────┐ ┌──────▼─────┐ │
│ │ HAOS Adapter │ │ MQTT Adapter │ │ Zigbee │ │
│ │ (REST/WS) │ │ (Mosquitto) │ │ Adapter │ │
│ └────────────────┘ └──────────────┘ └────────────┘ │
└──────────────────────────────────────────────────────┘
│ │ │
▼ ▼ ▼
Home Assistant MQTT Broker Zigbee2MQTT
(on HAOS or VM) (standalone) (Coordinator)
CloseClaw connects to Home Assistant via:
- Long-Lived Access Token (stored in local encrypted vault, see §5.4)
- WebSocket API for real-time state subscriptions
- REST API for entity control and history queries
homeassistant:
enabled: true
host: 192.168.1.50 # HA IP (LAN only)
port: 8123
token_vault_key: ha_token # Key in CloseClaw encrypted vault
tls: false # Enable if HA has self-signed cert
# What CloseClaw exposes back to HA
expose_as_integration: true
entity_prefix: closeclaw
# Entity sync — which HA entities CloseClaw monitors
watch_domains:
- light
- switch
- sensor
- binary_sensor
- climate
- media_player
- cover
# Automation mode
automation:
allow_create: true # CloseClaw can write HA automations
allow_delete: false # Safety: don't delete existing ones
namespace: closeclaw_ai # Automation label prefix# closeclaw/integrations/homeassistant/adapter.py
import aiohttp
import json
class HomeAssistantAdapter:
def __init__(self, config: HAConfig, vault: SecretVault):
self.base_url = f"http://{config.host}:{config.port}"
self.token = vault.get(config.token_vault_key)
self.ws = None
self._entity_cache: dict[str, dict] = {}
async def connect(self):
"""Establish WebSocket connection and subscribe to state changes."""
self.ws = await aiohttp.ClientSession().ws_connect(
f"{self.base_url}/api/websocket"
)
await self._authenticate()
await self._subscribe_state_changes()
await self._load_all_entities()
async def call_service(self, domain: str, service: str,
entity_id: str, **kwargs):
"""Execute a HA service call (e.g., light.turn_on)."""
payload = {
"type": "call_service",
"domain": domain,
"service": service,
"service_data": {"entity_id": entity_id, **kwargs}
}
await self.ws.send_json(payload)
async def get_state(self, entity_id: str) -> dict:
"""Get current state from cache or HA."""
if entity_id in self._entity_cache:
return self._entity_cache[entity_id]
async with aiohttp.ClientSession() as s:
r = await s.get(
f"{self.base_url}/api/states/{entity_id}",
headers={"Authorization": f"Bearer {self.token}"}
)
return await r.json()
def get_home_summary(self) -> dict:
"""Generate a structured home state for LLM context."""
return {
"lights_on": [e for e, s in self._entity_cache.items()
if e.startswith("light.") and s["state"] == "on"],
"doors_open": [e for e, s in self._entity_cache.items()
if e.startswith("binary_sensor.")
and "door" in e and s["state"] == "on"],
"temperature": {e: s["attributes"].get("temperature")
for e, s in self._entity_cache.items()
if e.startswith("climate.")},
"occupied_rooms": self._infer_occupancy(),
}CloseClaw's LLM layer converts natural language commands into structured HA service calls:
# closeclaw/skills/home_control.py
SYSTEM_PROMPT = """You are CloseClaw's home control brain.
You have access to these Home Assistant entities:
{entity_list}
Current home state:
{home_summary}
Convert the user's command into a JSON action list.
Use ONLY entities that exist. Never fabricate entity IDs.
Output format: [{"domain": "light", "service": "turn_on", "entity_id": "light.living_room", "kwargs": {"brightness": 200}}]
If you cannot fulfill the request, explain why."""
async def execute_home_command(utterance: str, ha: HomeAssistantAdapter):
entity_list = "\n".join(ha.list_entities())
home_summary = json.dumps(ha.get_home_summary(), indent=2)
response = await local_llm.complete(
system=SYSTEM_PROMPT.format(
entity_list=entity_list,
home_summary=home_summary
),
prompt=utterance,
model="medium" # Hub's capable model
)
actions = json.loads(response)
for action in actions:
await ha.call_service(
action["domain"], action["service"],
action["entity_id"], **action.get("kwargs", {})
)CloseClaw ships with an embedded Mosquitto broker (or connects to an existing one). The MQTT adapter subscribes to device topics and translates messages into the CloseClaw event bus.
# config/integrations/mqtt.yaml
mqtt:
mode: embedded # "embedded" | "external"
# If embedded:
broker:
port: 1883
tls_port: 8883
persistence: true
persistence_path: /var/lib/closeclaw/mqtt
# If external:
external:
host: 192.168.1.60
port: 8883
client_id: closeclaw-hub
# Topic mappings to CloseClaw events
subscriptions:
- topic: "zigbee2mqtt/+"
handler: zigbee2mqtt
- topic: "zwave2mqtt/+"
handler: zwave
- topic: "tasmota/+/SENSOR"
handler: tasmota_sensor
- topic: "shellies/+"
handler: shellyZigbee2MQTT publishes device states and receives commands over MQTT. CloseClaw's Zigbee adapter translates between Zigbee's device model and CloseClaw's unified sensor/actuator API (see §4).
# closeclaw/integrations/zigbee2mqtt/adapter.py
ZIGBEE_DEVICE_MAP = {
"TRADFRI bulb": {"type": "light", "capabilities": ["on_off", "brightness", "color_temp"]},
"SNZB-02": {"type": "sensor", "capabilities": ["temperature", "humidity"]},
"E1743": {"type": "switch", "capabilities": ["on_off"]},
# ... extensible via config/zigbee_devices.yaml
}
class Zigbee2MQTTAdapter:
def __init__(self, mqtt_client: MQTTClient):
self.mqtt = mqtt_client
self.devices: dict[str, ZigbeeDevice] = {}
async def on_message(self, topic: str, payload: bytes):
# zigbee2mqtt/<friendly_name> or zigbee2mqtt/<friendly_name>/availability
parts = topic.split("/")
device_name = parts[1]
if len(parts) == 2: # State update
state = json.loads(payload)
device = self.devices.get(device_name)
if device:
reading = self._to_sensor_reading(device, state)
EventBus.emit(EventType.SENSOR_TEMP, reading)
def set_state(self, device_name: str, **kwargs):
"""Send command to Zigbee device."""
topic = f"zigbee2mqtt/{device_name}/set"
self.mqtt.publish(topic, json.dumps(kwargs))
def _to_sensor_reading(self, device: ZigbeeDevice,
raw: dict) -> SensorReading:
return SensorReading(
device_id=device.id,
device_type=device.type,
values=raw,
timestamp=time.time(),
source="zigbee2mqtt"
)Z-Wave support is provided via zwave-js-ui (formerly Z-Wave to MQTT). CloseClaw connects via its MQTT interface using the same adapter pattern as Zigbee2MQTT.
# config/integrations/zwave.yaml
zwave:
enabled: true
zwavejs_ui_host: 192.168.1.51
mqtt_prefix: "zwave"
# Node ID to friendly name mapping (auto-populated on first connect)
node_aliases:
"6": "front_door_lock"
"12": "garage_door"
"18": "basement_thermostat"CloseClaw exposes a bidirectional integration: beyond consuming HA data, it registers itself as a custom component in HA, exposing:
- CloseClaw nodes as HA
deviceentities - Inference requests as HA
conversationactions - Local AI responses visible in HA's UI
- CloseClaw-managed automations visible in HA's automation list
The HA custom component lives at custom_components/closeclaw/ and communicates with the hub via local gRPC (never HTTP to the internet).
The voice pipeline runs entirely on-device, including on a Raspberry Pi 4 (2GB+). No cloud STT, no cloud TTS, no cloud NLU. The pipeline is:
Microphone → Wake Word → VAD → STT (Whisper.cpp) → NLU (Local LLM) → TTS (Piper) → Speaker
Each stage is independently configurable and can run on different mesh nodes if resources are constrained.
openWakeWord runs first — always listening, ultra-low CPU:
# closeclaw/voice/wake_word.py
import openwakeword
from openwakeword.model import Model
class WakeWordDetector:
# Default models — user can add custom wake words
DEFAULT_MODELS = [
"hey_jarvis", # fallback generic
"hey_closeclaw", # custom-trained (see training guide)
]
def __init__(self, config: VoiceConfig):
self.model = Model(
wakeword_models=config.wake_words or self.DEFAULT_MODELS,
enable_speex_noise_suppression=True,
vad_threshold=0.5,
noise_suppression_level=2
)
self.threshold = config.wake_word_threshold or 0.7
def process_chunk(self, audio_chunk: bytes) -> bool:
prediction = self.model.predict_chunk(audio_chunk)
return any(score > self.threshold for score in prediction.values())Pi Zero 2W performance: openWakeWord consumes ~15-25% CPU on a Pi Zero 2W. Deployable as a dedicated always-on wake word listener node.
Custom wake words: CloseClaw includes a training script (tools/train_wake_word.py) that uses openWakeWord's training pipeline to create custom wake words from ~30 positive samples, all locally.
After wake word triggers, VAD determines when the user has finished speaking:
# closeclaw/voice/vad.py
# Uses Silero VAD (ONNX, runs on CPU, Pi-compatible)
class SileroVAD:
def __init__(self):
import onnxruntime as ort
model_path = MODELS_DIR / "silero_vad.onnx"
self.session = ort.InferenceSession(str(model_path))
self.threshold = 0.5
self.silence_duration_ms = 800 # End utterance after 800ms silence
def is_speech(self, audio_chunk: np.ndarray) -> bool:
# 16kHz, 16-bit, mono required
tensor = audio_chunk.astype(np.float32) / 32768.0
out = self.session.run(None, {"input": tensor[None]})[0]
return float(out) > self.threshold
def collect_utterance(self, audio_stream) -> bytes:
"""Collect audio until end of speech detected."""
utterance_frames = []
silence_frames = 0
for frame in audio_stream:
if self.is_speech(frame):
utterance_frames.append(frame)
silence_frames = 0
else:
silence_frames += 1
if utterance_frames: # Only count silence after speech
utterance_frames.append(frame)
if silence_frames > (self.silence_duration_ms / 30):
break # End of utterance
return b"".join(utterance_frames)whisper.cpp handles STT — extremely Pi-friendly via quantized models:
# config/voice.yaml
stt:
backend: whisper_cpp
model: base.en # "tiny.en" for Pi Zero, "base.en" for Pi 4, "small.en" for hub
quantization: q5_k_m # Quantized for ARM
language: en
threads: 4 # Match to Pi CPU core count
beam_size: 1 # Reduce for speed on slow devices
# Model performance reference:
# tiny.en q5: Pi Zero 2W ~2-3x realtime (fast, less accurate)
# base.en q5: Pi 4 2GB ~1.5x realtime (good balance)
# small.en q5: Pi 4 4GB ~1x realtime (accurate)
# medium.en: Hub/NUC ~0.5x realtime (best)# closeclaw/voice/stt.py
import subprocess
import tempfile
import wave
class WhisperCppSTT:
def __init__(self, config: STTConfig):
self.binary = CLOSECLAW_BIN / "whisper-cpp"
self.model = MODELS_DIR / f"ggml-{config.model}.bin"
self.threads = config.threads
self.language = config.language
async def transcribe(self, audio_bytes: bytes) -> str:
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
self._write_wav(f.name, audio_bytes)
result = await asyncio.create_subprocess_exec(
str(self.binary),
"-m", str(self.model),
"-f", f.name,
"-l", self.language,
"-t", str(self.threads),
"--no-timestamps",
"--output-txt",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL
)
stdout, _ = await result.communicate()
return stdout.decode().strip()
def _write_wav(self, path: str, pcm_bytes: bytes):
with wave.open(path, "wb") as w:
w.setnchannels(1)
w.setsampwidth(2) # 16-bit
w.setframerate(16000) # 16kHz
w.writeframes(pcm_bytes)Piper is a fast, high-quality neural TTS that runs well on Pi:
# config/voice.yaml
tts:
backend: piper
voice: en_US-lessac-medium # VITS neural voice
speaker_id: 0
noise_scale: 0.667
length_scale: 1.0
# Voice quality vs speed tradeoff:
# low: Pi Zero 2W usable (~0.8x realtime)
# medium: Pi 4 good speed (~2x realtime)
# high: Hub/NUC quality (~5x realtime)
# Fallback if Piper unavailable:
fallback: espeak
espeak_voice: en-us# closeclaw/voice/tts.py
import subprocess
import io
class PiperTTS:
def __init__(self, config: TTSConfig):
self.binary = CLOSECLAW_BIN / "piper"
self.model = MODELS_DIR / "tts" / f"{config.voice}.onnx"
self.config_path = MODELS_DIR / "tts" / f"{config.voice}.onnx.json"
async def synthesize(self, text: str) -> bytes:
"""Returns raw PCM audio (16kHz, 16-bit, mono)."""
process = await asyncio.create_subprocess_exec(
str(self.binary),
"--model", str(self.model),
"--config", str(self.config_path),
"--output-raw",
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL
)
audio, _ = await process.communicate(text.encode())
return audio
async def speak(self, text: str, output_device: str = "default"):
"""Synthesize and play immediately."""
audio = await self.synthesize(text)
await self._play_audio(audio, output_device)
async def _play_audio(self, pcm: bytes, device: str):
await asyncio.create_subprocess_exec(
"aplay", "-r", "22050", "-f", "S16_LE", "-c", "1",
"-D", device, "-",
stdin=asyncio.subprocess.PIPE
).communicate(pcm)# closeclaw/voice/pipeline.py
class VoicePipeline:
def __init__(self, config: VoiceConfig, mesh: MeshClient):
self.wake_word = WakeWordDetector(config)
self.vad = SileroVAD()
self.stt = WhisperCppSTT(config.stt)
self.tts = PiperTTS(config.tts)
self.mesh = mesh
self.audio_in = AudioCapture(config.microphone_device)
async def run(self):
"""Main voice loop — runs forever."""
async for chunk in self.audio_in.stream():
if self.wake_word.process_chunk(chunk):
await self._handle_utterance()
async def _handle_utterance(self):
# Signal wake word detected
await self._play_ding()
# Collect full utterance
utterance_audio = self.vad.collect_utterance(
self.audio_in.stream_unbuffered()
)
# Transcribe locally
text = await self.stt.transcribe(utterance_audio)
if not text.strip():
return
EventBus.emit(EventType.VOICE_UTTERANCE, {
"text": text, "node": NODE_ID
})
# Send to inference (may be delegated to hub)
response = await self.mesh.infer(
prompt=text,
context={"source": "voice", "home_state": ha_adapter.get_home_summary()},
model="medium"
)
# Speak response locally
await self.tts.speak(response.text)
EventBus.emit(EventType.VOICE_RESPONSE, {"text": response.text})
async def _play_ding(self):
"""Short chime to indicate wake word detected."""
audio = load_wav(ASSETS_DIR / "wake_chime.wav")
await self._play_raw(audio)Each room can have a dedicated voice node (Pi Zero 2W for wake word, Pi 4 for full STT). The mesh coordinates:
- Wake word: all nodes listen simultaneously
- Utterance capture: nearest node (detected by audio level) takes ownership
- STT: performed on the capturing node or delegated to hub if underpowered
- Response: played back on the capturing node's speaker
Room ownership is determined by which node first exceeds the VAD threshold after a shared wake word event.
CloseClaw presents a single, consistent API for all physical I/O — whether it's a GPIO pin on a Pi, a servo on a robot, a Zigbee light, or an IP camera. This abstraction allows automations and the AI layer to interact with hardware identically regardless of the underlying protocol.
# closeclaw/devices/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
class DeviceType(Enum):
SENSOR = "sensor"
ACTUATOR = "actuator"
CAMERA = "camera"
MICROPHONE = "microphone"
SPEAKER = "speaker"
GPIO_IN = "gpio_in"
GPIO_OUT = "gpio_out"
SERVO = "servo"
RELAY = "relay"
MOTOR = "motor"
@dataclass
class DeviceCapabilities:
can_read: bool = False
can_write: bool = False
can_stream: bool = False
supports_events: bool = False
units: str = "" # "celsius", "percent", "boolean", etc.
value_range: tuple = None # (min, max) if applicable
class ClawDevice(ABC):
"""Base class for ALL CloseClaw devices."""
def __init__(self, device_id: str, name: str, node_id: str):
self.device_id = device_id
self.name = name
self.node_id = node_id
self.capabilities = DeviceCapabilities()
@abstractmethod
async def read(self) -> "SensorReading":
"""Read current device state/value."""
...
@abstractmethod
async def write(self, value) -> bool:
"""Set device state/value. Returns success."""
...
async def stream(self):
"""Async generator yielding readings. Override for streaming devices."""
while True:
yield await self.read()
await asyncio.sleep(self.poll_interval)
def to_dict(self) -> dict:
return {
"id": self.device_id,
"name": self.name,
"node": self.node_id,
"type": self.__class__.__name__,
"capabilities": vars(self.capabilities)
}# closeclaw/devices/gpio.py
import RPi.GPIO as GPIO
class GPIOInput(ClawDevice):
def __init__(self, device_id: str, name: str, pin: int,
pull_up: bool = True, invert: bool = False):
super().__init__(device_id, name, NODE_ID)
self.pin = pin
self.invert = invert
self.capabilities = DeviceCapabilities(
can_read=True, supports_events=True,
units="boolean"
)
GPIO.setup(pin, GPIO.IN,
pull_up_down=GPIO.PUD_UP if pull_up else GPIO.PUD_DOWN)
GPIO.add_event_detect(pin, GPIO.BOTH,
callback=self._on_change, bouncetime=50)
async def read(self) -> SensorReading:
raw = GPIO.input(self.pin)
value = not raw if self.invert else bool(raw)
return SensorReading(
device_id=self.device_id,
value=value,
unit="boolean",
timestamp=time.time()
)
def _on_change(self, channel):
reading = asyncio.run(self.read())
EventBus.emit(EventType.SENSOR_DOOR, reading)# closeclaw/devices/camera.py
import cv2
class Camera(ClawDevice):
def __init__(self, device_id: str, name: str,
source, # int (USB index), str (RTSP URL), or "csi"
resolution: tuple = (640, 480), fps: int = 10):
super().__init__(device_id, name, NODE_ID)
self.source = source
self.resolution = resolution
self.fps = fps
self.capabilities = DeviceCapabilities(
can_read=True, can_stream=True, units="image/jpeg"
)
self._cap = None
async def read(self) -> SensorReading:
"""Capture single frame."""
if self._cap is None:
self._open()
ret, frame = self._cap.read()
if not ret:
raise DeviceError(f"Camera {self.device_id} failed to capture")
_, jpeg = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 80])
return SensorReading(
device_id=self.device_id,
value=jpeg.tobytes(),
unit="image/jpeg",
metadata={"width": frame.shape[1], "height": frame.shape[0]},
timestamp=time.time()
)
async def stream(self):
"""Async generator for continuous frames."""
self._open()
try:
while True:
yield await self.read()
await asyncio.sleep(1.0 / self.fps)
finally:
self._cap.release()
def _open(self):
if self.source == "csi":
# Pi Camera v2/v3 via libcamera
pipeline = (
"libcamerasrc ! video/x-raw,format=RGB,width=640,height=480 ! "
"videoconvert ! appsink"
)
self._cap = cv2.VideoCapture(pipeline, cv2.CAP_GSTREAMER)
else:
self._cap = cv2.VideoCapture(self.source)
self._cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.resolution[0])
self._cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.resolution[1])# closeclaw/devices/microphone.py
import sounddevice as sd
import numpy as np
class Microphone(ClawDevice):
def __init__(self, device_id: str, name: str,
device_index: int = None,
sample_rate: int = 16000,
chunk_size: int = 512):
super().__init__(device_id, name, NODE_ID)
self.device_index = device_index
self.sample_rate = sample_rate
self.chunk_size = chunk_size
self.capabilities = DeviceCapabilities(
can_read=True, can_stream=True, units="audio/pcm16"
)
async def read(self) -> SensorReading:
"""Capture one chunk of audio."""
audio = sd.rec(self.chunk_size, samplerate=self.sample_rate,
channels=1, dtype="int16", device=self.device_index)
sd.wait()
return SensorReading(
device_id=self.device_id,
value=audio.tobytes(),
unit="audio/pcm16",
metadata={"sample_rate": self.sample_rate, "channels": 1},
timestamp=time.time()
)
async def stream(self):
"""Continuous audio stream as async generator."""
queue = asyncio.Queue()
def callback(indata, frames, time_info, status):
queue.put_nowait(indata.copy().tobytes())
with sd.InputStream(samplerate=self.sample_rate, channels=1,
dtype="int16", device=self.device_index,
blocksize=self.chunk_size, callback=callback):
while True:
chunk = await queue.get()
yield SensorReading(
device_id=self.device_id,
value=chunk,
unit="audio/pcm16",
timestamp=time.time()
)# closeclaw/devices/relay.py
import RPi.GPIO as GPIO
class Relay(ClawDevice):
"""Controls a relay or any GPIO output (LED, buzzer, solenoid)."""
def __init__(self, device_id: str, name: str, pin: int,
active_high: bool = True, initial_state: bool = False):
super().__init__(device_id, name, NODE_ID)
self.pin = pin
self.active_high = active_high
self._state = initial_state
self.capabilities = DeviceCapabilities(
can_read=True, can_write=True, units="boolean"
)
GPIO.setup(pin, GPIO.OUT)
self._apply(initial_state)
async def read(self) -> SensorReading:
return SensorReading(
device_id=self.device_id, value=self._state,
unit="boolean", timestamp=time.time()
)
async def write(self, value: bool) -> bool:
self._state = bool(value)
self._apply(self._state)
EventBus.emit(EventType.DEVICE_STATE, {
"device_id": self.device_id, "state": self._state
})
return True
def _apply(self, state: bool):
gpio_val = state if self.active_high else not state
GPIO.output(self.pin, GPIO.HIGH if gpio_val else GPIO.LOW)# closeclaw/devices/servo.py
from gpiozero import Servo as GZServo
from gpiozero.pins.pigpio import PiGPIOFactory
class ServoMotor(ClawDevice):
"""PWM servo (0-180 degrees)."""
def __init__(self, device_id: str, name: str, pin: int,
min_pulse: float = 0.5, max_pulse: float = 2.5):
super().__init__(device_id, name, NODE_ID)
factory = PiGPIOFactory() # pigpiod for precise PWM
self._servo = GZServo(pin, min_pulse_width=min_pulse/1000,
max_pulse_width=max_pulse/1000,
pin_factory=factory)
self._angle = 90.0
self.capabilities = DeviceCapabilities(
can_read=True, can_write=True,
units="degrees", value_range=(0, 180)
)
async def read(self) -> SensorReading:
return SensorReading(
device_id=self.device_id, value=self._angle,
unit="degrees", timestamp=time.time()
)
async def write(self, angle: float) -> bool:
"""Set servo angle (0-180 degrees)."""
angle = max(0, min(180, float(angle)))
# gpiozero servo value: -1 (0°) to 1 (180°)
normalized = (angle / 90.0) - 1.0
self._servo.value = normalized
self._angle = angle
return TrueThe Device Registry is the single source of truth for all hardware in the mesh:
# config/devices.yaml — auto-generated + user-editable
devices:
- id: cam-front-door
type: camera
name: "Front Door Camera"
node: rpi-zero-01
driver: camera
config:
source: "rtsp://192.168.1.201/stream1"
resolution: [1280, 720]
fps: 5
- id: relay-porch-light
type: relay
name: "Porch Light Relay"
node: rpi-zero-02
driver: gpio_output
config:
pin: 17
active_high: true
- id: servo-camera-pan
type: servo
name: "Camera Pan Servo"
node: robot-01
driver: servo
config:
pin: 12
min_pulse: 0.5
max_pulse: 2.5
- id: zigbee-living-room-temp
type: sensor
name: "Living Room Temperature"
node: hub
driver: zigbee2mqtt
config:
friendly_name: "living_room_sensor"CloseClaw's security model assumes:
- LAN is untrusted — an attacker may have access to the local network (compromised device, rogue guest)
- No internet — cloud certificate authorities and external PKI services are unavailable
- Physical security — devices may be in accessible locations (outdoor cameras, door sensors)
- User is trusted — the human with physical access to the hub is the root of trust
Non-goals:
- Protection against a malicious hub owner (root of trust must start somewhere)
- Resistance to physical device teardown (hardware tamper protection is out of scope for v1)
On first boot, the CloseClaw Hub generates a local Certificate Authority:
# Automatic on hub first-boot:
# closeclaw/security/pki.py
class LocalCA:
CA_KEY_BITS = 4096
DEVICE_KEY_BITS = 2048
CA_VALIDITY_DAYS = 3650 # 10 years
DEVICE_VALIDITY_DAYS = 365
def __init__(self, ca_dir: Path):
self.ca_dir = ca_dir
self.ca_key_path = ca_dir / "ca.key"
self.ca_cert_path = ca_dir / "ca.crt"
def initialize(self):
"""Generate CA key pair and self-signed certificate."""
# Generate CA private key (RSA 4096 or Ed25519)
ca_key = crypto.PKey()
ca_key.generate_key(crypto.TYPE_RSA, self.CA_KEY_BITS)
# Self-signed CA certificate
ca_cert = crypto.X509()
ca_cert.get_subject().CN = "CloseClaw Mesh CA"
ca_cert.get_subject().O = f"CloseClaw-{self._mesh_id()}"
ca_cert.set_issuer(ca_cert.get_subject())
ca_cert.gmtime_adj_notBefore(0)
ca_cert.gmtime_adj_notAfter(86400 * self.CA_VALIDITY_DAYS)
ca_cert.add_extensions([
crypto.X509Extension(b"basicConstraints", True, b"CA:TRUE"),
crypto.X509Extension(b"keyUsage", True, b"keyCertSign, cRLSign"),
])
ca_cert.sign(ca_key, "sha256")
# Store encrypted (passphrase from vault)
self._save_key(ca_key)
self._save_cert(ca_cert)
def issue_device_cert(self, device_id: str,
csr_pem: bytes) -> bytes:
"""Sign a device's CSR and return the certificate."""
csr = crypto.load_certificate_request(crypto.FILETYPE_PEM, csr_pem)
cert = crypto.X509()
cert.get_subject().CN = device_id
cert.set_issuer(self.ca_cert.get_subject())
cert.set_pubkey(csr.get_pubkey())
cert.gmtime_adj_notBefore(0)
cert.gmtime_adj_notAfter(86400 * self.DEVICE_VALIDITY_DAYS)
cert.add_extensions([
crypto.X509Extension(b"subjectAltName", False,
f"DNS:{device_id}.closeclaw.local".encode()),
])
cert.sign(self.ca_key, "sha256")
return crypto.dump_certificate(crypto.FILETYPE_PEM, cert)CA storage: The CA private key is stored in the hub's encrypted vault (see §5.4), never on disk in plaintext.
New devices must be explicitly paired before they can join the mesh. Two pairing methods:
┌─────────────────────────────────────────────────────┐
│ CloseClaw Hub generates pairing token: │
│ │
│ token = { │
│ "mesh_id": "a3f7b21c", │
│ "hub_ip": "192.168.1.100", │
│ "hub_pubkey": "sha256:b4e9...", │
│ "pairing_token": "<random 32-byte hex>", │
│ "expires_at": 1712248800 │
│ } │
│ │
│ Token encoded as QR code, displayed in hub UI. │
│ New device scans QR → sends CSR + token to hub │
│ Hub validates token, signs CSR, device is paired │
└─────────────────────────────────────────────────────┘
# closeclaw/security/pairing.py
class PairingManager:
TOKEN_EXPIRY_SECONDS = 300 # 5 minute window
def generate_pairing_token(self) -> PairingToken:
token = secrets.token_hex(32)
expires_at = int(time.time()) + self.TOKEN_EXPIRY_SECONDS
pt = PairingToken(
mesh_id=self.mesh_id,
hub_ip=self.hub_ip,
hub_pubkey=self.hub_pubkey_fingerprint,
token=token,
expires_at=expires_at
)
# Store pending token (in-memory only, no disk)
self._pending_tokens[token] = pt
# Display QR code
qr_data = json.dumps(pt.to_dict())
self._display_qr(qr_data)
return pt
async def complete_pairing(self, device_csr: bytes,
token: str, device_name: str) -> bytes:
"""Called when new device sends its CSR + pairing token."""
if token not in self._pending_tokens:
raise PairingError("Invalid or expired pairing token")
pt = self._pending_tokens[token]
if time.time() > pt.expires_at:
del self._pending_tokens[token]
raise PairingError("Pairing token expired")
# Issue certificate
cert_pem = self.pki.issue_device_cert(device_name, device_csr)
# Register device
self.device_registry.register(DeviceRecord(
id=device_name,
cert_fingerprint=self._fingerprint(cert_pem),
paired_at=int(time.time())
))
del self._pending_tokens[token]
return cert_pemFor devices without screens (sensors, relay boards, Pi Zero):
1. User presses PAIR button on hub (or triggers via CloseClaw UI)
2. Hub enters 60-second pairing window — broadcasts beacon on UDP 7701
3. New device boots and sends discovery packet: {"type": "pair_request", "device_id": "<mac-based-id>", "csr": "<base64>"}
4. Hub receives, displays device ID in UI/terminal, prompts for user confirmation
5. User approves → hub signs CSR, sends cert back to device
6. Device stores cert in encrypted flash (see §5.4), pairing complete
Sensitive data (CA key, HA token, user data) is encrypted at rest using age (a modern encryption tool):
# closeclaw/security/vault.py
import subprocess
import json
class SecretVault:
"""Encrypted key-value store backed by age encryption."""
def __init__(self, vault_path: Path, master_key_path: Path):
self.vault_path = vault_path
self.master_key_path = master_key_path
self._cache: dict = {}
def get(self, key: str) -> str:
if key in self._cache:
return self._cache[key]
data = self._decrypt_vault()
return data.get(key)
def set(self, key: str, value: str):
data = self._decrypt_vault()
data[key] = value
self._encrypt_vault(data)
self._cache[key] = value
def _decrypt_vault(self) -> dict:
result = subprocess.run(
["age", "--decrypt", "-i", str(self.master_key_path),
str(self.vault_path)],
capture_output=True
)
return json.loads(result.stdout)
def _encrypt_vault(self, data: dict):
plaintext = json.dumps(data).encode()
result = subprocess.run(
["age", "-r", self._get_recipient(),
"-o", str(self.vault_path)],
input=plaintext, capture_output=True
)Master key derivation: On hub, master key is derived from a combination of:
- Hardware-unique identifier (Pi serial number, CPU ID)
- User-set passphrase (optional, prompted on hub boot if set)
- TPM binding (future v2 feature)
Every gRPC connection between nodes uses mutual TLS:
# closeclaw/mesh/client.py
import grpc
def create_secure_channel(peer_host: str, peer_port: int) -> grpc.Channel:
creds = grpc.ssl_channel_credentials(
root_certificates=load_ca_cert(), # Validate peer's cert
private_key=load_node_key(), # Our private key
certificate_chain=load_node_cert() # Our certificate
)
return grpc.secure_channel(f"{peer_host}:{peer_port}", creds)
def create_secure_server(port: int) -> grpc.Server:
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
creds = grpc.ssl_server_credentials(
[(load_node_key(), load_node_cert())],
root_certificates=load_ca_cert(),
require_client_auth=True # mTLS — require client cert
)
server.add_secure_port(f"[::]:{port}", creds)
return serverEvery mesh event carries an Ed25519 signature:
# closeclaw/security/signing.py
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey, Ed25519PublicKey
)
class EventSigner:
def __init__(self, private_key: Ed25519PrivateKey):
self.private_key = private_key
def sign_event(self, event: NodeEvent) -> bytes:
"""Sign the canonical form of an event."""
payload = self._canonical_bytes(event)
return self.private_key.sign(payload)
def verify_event(self, event: NodeEvent,
sender_pubkey: Ed25519PublicKey) -> bool:
try:
payload = self._canonical_bytes(event)
sender_pubkey.verify(event.signature, payload)
return True
except Exception:
return False
def _canonical_bytes(self, event: NodeEvent) -> bytes:
# Deterministic serialization (sorted keys)
return json.dumps({
"event_id": event.event_id,
"source_node": event.source_node,
"event_type": event.event_type,
"payload_hash": hashlib.sha256(event.payload).hexdigest(),
"timestamp_ms": event.timestamp_ms,
}, sort_keys=True).encode()A Pi Zero 2W has ~512MB RAM. A modern LLM needs 4-8GB minimum. But the mesh has heterogeneous compute:
┌─────────────────────────────────────────────────────────┐
│ Mesh Compute Hierarchy │
│ │
│ Hub (NUC/NAS/x86) ████████████████ Large models │
│ 8-64GB RAM, optional GPU gemma2:27b │
│ mistral:7b │
│ │
│ Pi 4 (4-8GB) ████████ Medium models │
│ Edge nodes, robots gemma2:2b-q8 │
│ phi3:mini │
│ │
│ Pi 4 (2GB) ████ Small models │
│ Voice nodes qwen2:0.5b │
│ tinyllama │
│ │
│ Pi Zero 2W (512MB) ██ Inference-only │
│ Sensors, cameras (no local LLM) │
│ Wake word only │
└─────────────────────────────────────────────────────────┘
The Inference Router examines each request and routes it to the most capable node that can handle it within acceptable latency:
# closeclaw/inference/router.py
@dataclass
class InferenceRequest:
prompt: str
context: dict
model_hint: str # "tiny", "small", "medium", "large"
max_latency_ms: int = 5000
requester_node: str = ""
class InferenceRouter:
def __init__(self, mesh: MeshClient, local_engine: LocalInferenceEngine):
self.mesh = mesh
self.local = local_engine
self._node_capabilities: dict[str, NodeCapability] = {}
async def route(self, request: InferenceRequest) -> InferenceResponse:
target = self._select_target(request)
if target == NODE_ID:
return await self.local.infer(request)
else:
return await self.mesh.delegate_inference(target, request)
def _select_target(self, request: InferenceRequest) -> str:
"""Select optimal inference node."""
suitable = [
(node_id, cap)
for node_id, cap in self._node_capabilities.items()
if cap.can_handle(request.model_hint)
and cap.estimated_latency_ms(request) < request.max_latency_ms
]
if not suitable:
# Fall back to hub regardless of latency
return self._find_hub()
# Prefer closest capable node (minimize network hop latency)
# Secondary sort by current load
suitable.sort(key=lambda x: (x[1].hop_count, x[1].current_load))
return suitable[0][0]
def _find_hub(self) -> str:
"""Find the primary hub node."""
hubs = [nid for nid, cap in self._node_capabilities.items()
if cap.role == "hub"]
return hubs[0] if hubs else NODE_ID # Fallback to self# closeclaw/inference/model_selector.py
MODEL_TIERS = {
"tiny": [
"tinyllama:1.1b-q4", # ~600MB, Pi 4 2GB
"qwen2:0.5b-q4", # ~400MB, Pi Zero 2W
"phi3:mini-q4", # ~2GB, Pi 4 4GB
],
"small": [
"gemma2:2b-q8", # ~2.5GB, Pi 4 4GB
"phi3:mini-fp16", # ~4GB, Pi 4 8GB
"llama3.2:3b-q6", # ~3GB, Pi 4 4GB
],
"medium": [
"gemma2:9b-q5", # ~6GB, NUC/Hub
"mistral:7b-q8", # ~8GB, Hub
"llama3.1:8b-q5", # ~6GB, Hub
],
"large": [
"gemma2:27b-q4", # ~18GB, Hub with 32GB+ RAM
"mixtral:8x7b-q3", # ~24GB, Hub with GPU
"llama3.1:70b-q4", # ~40GB, Hub with 64GB RAM
]
}
def select_model_for_task(task_type: str, available_models: list[str]) -> str:
"""Match task complexity to appropriate model."""
task_complexity = {
"wake_word_response": "tiny",
"simple_command": "small",
"home_automation": "small",
"conversation": "medium",
"complex_reasoning": "medium",
"code_generation": "large",
"multimodal": "medium", # + vision adapter
}
required_tier = task_complexity.get(task_type, "small")
candidates = MODEL_TIERS[required_tier]
for model in candidates:
if model in available_models:
return model
# Fall back to best available
for tier in ["medium", "small", "tiny"]:
for model in MODEL_TIERS[tier]:
if model in available_models:
return model
raise NoModelAvailableError("No suitable model found in mesh")For large models that don't fit on a single device, CloseClaw implements tensor parallelism across mesh nodes using llama.cpp's RPC backend:
# config/inference/sharding.yaml
model_sharding:
enabled: true
shards:
# 27B model split across hub (24 layers) + Pi 4 (8 layers)
- model: "gemma2:27b-q4"
strategy: "layer_split"
nodes:
- node_id: hub-01
layers: 0-23 # First 24 transformer layers
device: cpu
offload_kv: true
- node_id: rpi4-01
layers: 24-31 # Final 8 layers + LM head
device: cpu
offload_kv: false# closeclaw/inference/sharding.py
# Based on llama.cpp's --rpc flag for distributed inference
class ShardedInferenceEngine:
def __init__(self, shard_config: ShardConfig, mesh: MeshClient):
self.config = shard_config
self.mesh = mesh
self._rpc_servers: dict[str, str] = {} # node_id -> rpc_url
async def setup_shards(self):
"""Start llama.cpp RPC server on each shard node."""
for shard in self.config.shards:
if shard.node_id == NODE_ID:
# Start local RPC server
self._start_local_rpc_server(shard)
else:
# Start remote RPC server via mesh command
await self.mesh.execute(shard.node_id, {
"command": "start_rpc_server",
"port": 50052,
"layers": shard.layers
})
self._rpc_servers[shard.node_id] = f"{shard.host}:50052"
def _start_local_rpc_server(self, shard: ShardDefinition):
"""Launch llama-rpc-server for this node's layers."""
import subprocess
subprocess.Popen([
str(CLOSECLAW_BIN / "llama-rpc-server"),
"--host", "0.0.0.0",
"--port", "50052",
])
async def infer(self, request: InferenceRequest) -> InferenceResponse:
"""Run inference across all shards."""
# Primary shard runs llama.cpp with --rpc pointing to other shards
rpc_args = ",".join(self._rpc_servers.values())
result = await asyncio.create_subprocess_exec(
str(CLOSECLAW_BIN / "llama-cli"),
"-m", str(self.model_path),
"--rpc", rpc_args,
"-n", str(request.max_tokens),
"-p", request.prompt,
stdout=asyncio.subprocess.PIPE
)
stdout, _ = await result.communicate()
return InferenceResponse(text=stdout.decode().strip())For latency-sensitive tasks (voice response), CloseClaw uses speculative decoding:
Edge Node (Pi 4):
- Runs small draft model (gemma2:2b-q4)
- Generates K candidate tokens per step
- Sends draft to hub for verification
Hub Node:
- Runs large target model (gemma2:27b-q4)
- Verifies edge's draft tokens in parallel
- Accepts/rejects, fills in divergences
- Returns verified tokens to edge
Result: Hub's quality at ~3-4x the speed
of running the large model alone
# closeclaw/inference/speculative.py
class SpeculativeDecoder:
def __init__(self, draft_model: LocalModel, hub: MeshPeer,
k: int = 8): # k = tokens to speculate per step
self.draft = draft_model
self.hub = hub
self.k = k
async def generate(self, prompt: str, max_tokens: int) -> str:
output_tokens = []
while len(output_tokens) < max_tokens:
# 1. Generate K draft tokens locally (fast)
drafts = await self.draft.generate_k(
prompt + "".join(output_tokens), k=self.k
)
# 2. Send drafts to hub for verification
verified = await self.hub.verify_speculative(
prompt=prompt + "".join(output_tokens),
draft_tokens=drafts
)
# 3. Accept verified prefix
output_tokens.extend(verified.accepted)
if verified.stop or len(output_tokens) >= max_tokens:
break
return "".join(output_tokens)CloseClaw maintains a local KV cache for frequent prompt prefixes (system prompts, home state context) to avoid recomputing attention layers:
# closeclaw/inference/cache.py
class PromptCache:
def __init__(self, max_entries: int = 16, max_tokens_per_entry: int = 512):
self.cache: dict[str, CacheEntry] = {}
self.max_entries = max_entries
def get_cached_kv(self, prefix: str) -> Optional[KVCache]:
"""Return cached KV state for a prompt prefix."""
key = hashlib.sha256(prefix.encode()).hexdigest()
if key in self.cache:
self.cache[key].hits += 1
return self.cache[key].kv_state
return None
def store(self, prefix: str, kv_state: KVCache):
if len(self.cache) >= self.max_entries:
# Evict least-recently-used entry
lru_key = min(self.cache.keys(),
key=lambda k: self.cache[k].last_used)
del self.cache[lru_key]
key = hashlib.sha256(prefix.encode()).hexdigest()
self.cache[key] = CacheEntry(
kv_state=kv_state, hits=0, last_used=time.time()
)
# Persistent system prompt prefix — cached on startup
HOME_CONTEXT_PREFIX = """You are CloseClaw, an offline AI assistant managing this home.
Current devices: {device_list}
Current state: {home_state}
Rules: Never suggest cloud services. All actions are local only."""Each node enforces inference resource limits to prevent starvation of sensor/actuator tasks:
# config/inference/resources.yaml
inference:
max_concurrent: 2
max_ram_fraction: 0.6 # Don't use more than 60% of available RAM
cpu_threads: 4
priority_queue:
- voice # Highest priority (user is waiting)
- automation # Medium (time-sensitive)
- background # Lowest (learning, indexing)
# Thermal throttling (Pi-specific)
throttle_above_celsius: 75
pause_above_celsius: 85# closeclaw/inference/resource_manager.py
class ResourceManager:
def __init__(self, config: ResourceConfig):
self.config = config
self._semaphore = asyncio.Semaphore(config.max_concurrent)
self._thermal_monitor = ThermalMonitor()
async def acquire(self, priority: str) -> AsyncContextManager:
# Block if at thermal limit
while self._thermal_monitor.temperature > self.config.pause_above:
await asyncio.sleep(5)
return self._semaphore
@contextlib.asynccontextmanager
async def inference_slot(self, priority: str):
async with await self.acquire(priority):
try:
yield
finally:
pass # Semaphore released automatically┌──────────────────────────────────┐
│ Raspberry Pi 4 (4GB) — Hub │
│ - CloseClaw Hub daemon │
│ - Ollama (gemma2:2b) │
│ - Mosquitto MQTT broker │
│ - HA integration bridge │
│ - Voice pipeline (Piper+Whis.) │
│ │
│ Connected devices: │
│ - USB microphone │
│ - USB speaker │
│ - Zigbee USB stick (CC2652) │
│ - Z-Wave USB stick (optional) │
└──────────────────────────────────┘
│ Zigbee mesh
├── TRADFRI lights
├── Aqara door sensors
└── SNZB-02 temp/humidity
┌─────────────────────┐
│ NUC Hub (32GB) │
│ gemma2:27b-q4 │
│ Hub orchestrator │
└─────────┬───────────┘
│ gRPC/mTLS (LAN)
┌─────────────────┼────────────────────┐
│ │ │
┌────────▼──────┐ ┌───────▼───────┐ ┌────────▼──────┐
│ Pi 4 Voice │ │ Pi 4 Robot │ │ Pi Zero Camera│
│ Bedroom │ │ Patrol Bot │ │ Front Door │
│ gemma2:2b-q4 │ │ gemma2:2b-q4 │ │ Wake word │
│ Piper TTS │ │ GPIO servos │ │ Motion detect│
│ Whisper STT │ │ Ultrasonic │ │ JPEG stream │
└───────────────┘ └───────────────┘ └───────────────┘
| Service | Port | Protocol | TLS |
|---|---|---|---|
| Mesh gRPC | 7700 | TCP | mTLS |
| MQTT | 1883 | TCP | No |
| MQTT TLS | 8883 | TCP | Client cert |
| mDNS discovery | 5353 | UDP/multicast | N/A |
| Pairing beacon | 7701 | UDP | N/A |
| llama.cpp RPC | 50052 | TCP | mTLS |
| Hub Web UI | 7780 | TCP | TLS |
- mDNS discovery (
_closeclaw._tcp.local) - Local PKI — CA generation, device cert issuance
- QR code pairing flow
- gRPC + mTLS mesh transport
- Device registry (YAML-based)
- GPIO input/output, relay, servo device drivers
- Secret vault (age-based)
- Home Assistant WebSocket + REST adapter
- MQTT embedded broker (Mosquitto)
- Zigbee2MQTT adapter
- Z-Wave (zwave-js-ui) adapter
- Camera device driver (USB + RTSP + CSI)
- Microphone + audio stream device driver
- openWakeWord integration
- Silero VAD
- Whisper.cpp STT (Pi-optimized, quantized models)
- Piper TTS (espeak fallback)
- Full voice pipeline orchestration
- Multi-room voice coordination
- Inference router (delegation to hub)
- Model tier selection
- Prompt cache
- Speculative decoding (Pi draft → Hub verify)
- Layer-split sharding (llama.cpp RPC)
- Thermal throttling + resource management
- Natural language → HA actions
- Event signing (Ed25519)
- mTLS certificate rotation
- Physical button pairing
- Hub election (multi-hub)
- HA custom integration (bidirectional)
- Custom wake word training pipeline
| Component | Library/Tool | License | Pi Zero 2W? |
|---|---|---|---|
| mDNS | python-zeroconf | LGPL | ✅ |
| mTLS transport | grpc[asyncio] | Apache 2.0 | ✅ |
| MQTT | paho-mqtt + mosquitto | EPL/OSI | ✅ |
| Wake word | openWakeWord | Apache 2.0 | ✅ (15-25% CPU) |
| VAD | silero-vad (ONNX) | MIT | ✅ |
| STT | whisper.cpp | MIT | ✅ (tiny.en q5) |
| TTS | piper | MIT | ✅ (low voice) |
| LLM inference | ollama / llama.cpp | MIT | ❌ (hub only) |
| GPIO | RPi.GPIO + gpiozero | LGPL | ✅ |
| Camera | OpenCV | Apache 2.0 | ✅ |
| Audio | sounddevice (PortAudio) | MIT | ✅ |
| Encryption | age | Apache 2.0 | ✅ |
| PKI | pyOpenSSL | Apache 2.0 | ✅ |
| Signing | cryptography (hazmat) | Apache 2.0 | ✅ |
This document is the canonical design for CloseClaw's Device Mesh & IoT Layer.
All design decisions prioritize: offline-first, security-by-default, Pi-compatibility.
The mesh is designed to be self-healing, encrypted, and air-gap safe.
CloseClaw: Your home's brain. Offline. Always.