docs/app_framework.md

ReachyMiniApp Framework

The Reachy Mini SDK provides a base class ReachyMiniApp for building applications. It handles lifecycle management, stop signals, and optionally serves a settings web UI.

Quick Reference

import threading
from reachy_mini import ReachyMini, ReachyMiniApp

class MyApp(ReachyMiniApp):
    custom_app_url: str | None = "http://0.0.0.0:8042"
    
    def run(self, reachy_mini: ReachyMini, stop_event: threading.Event):
        while not stop_event.is_set():
            # Your main loop here
            pass

if __name__ == "__main__":
    app = MyApp()
    app.wrapped_run()

Class Structure

ReachyMiniApp (Abstract Base Class)

Source: reachy_mini/apps/app.py

class ReachyMiniApp(ABC):
    custom_app_url: str | None = None      # URL for settings web UI
    dont_start_webserver: bool = False     # Skip web server startup
    
    def __init__(self) -> None:
        self.stop_event = threading.Event()
        self.error: str = ""
        self.settings_app: FastAPI | None = None
        # ...
    
    @abstractmethod
    def run(self, reachy_mini: ReachyMini, stop_event: threading.Event) -> None:
        """Main app logic - you implement this."""
        pass
    
    def wrapped_run(self, *args, **kwargs) -> None:
        """Wrapper that handles setup/teardown."""
        pass
    
    def stop(self) -> None:
        """Gracefully stop the app."""
        self.stop_event.set()

Key Attributes

custom_app_url

If set, the app will start a FastAPI web server at this URL.

custom_app_url: str | None = "http://0.0.0.0:8042"  # Serve at port 8042
custom_app_url: str | None = None                   # No web UI

settings_app

A FastAPI instance for adding custom API endpoints. Only available if custom_app_url is set.

@self.settings_app.post("/my_endpoint")
def my_handler():
    return {"status": "ok"}

stop_event

A threading.Event that gets set when the app should stop. Check this in your main loop:

while not stop_event.is_set():
    # Keep running
    time.sleep(0.01)

Static Files

The framework automatically serves static files from <your_app>/static/:

reachy_mini_metronome/
├── main.py              # Your app code
└── static/
    ├── index.html       # Served at /
    ├── style.css        # Served at /static/style.css
    └── script.js        # Served at /static/script.js

Source: reachy_mini/apps/app.py

static_dir = self._get_instance_path().parent / "static"
if static_dir.exists():
    self.settings_app.mount("/static", StaticFiles(directory=static_dir), name="static")
    
    index_file = static_dir / "index.html"
    if index_file.exists():
        @self.settings_app.get("/")
        async def index() -> FileResponse:
            return FileResponse(index_file)

Adding API Endpoints

You can add FastAPI endpoints in your run() method:

def run(self, reachy_mini: ReachyMini, stop_event: threading.Event):
    # State variables
    bpm = 120
    is_running = False
    
    # Define endpoints
    @self.settings_app.post("/bpm")
    def update_bpm(data: dict):
        nonlocal bpm
        bpm = data.get("bpm", bpm)
        return {"bpm": bpm}
    
    @self.settings_app.post("/start")
    def start():
        nonlocal is_running
        is_running = True
        return {"running": True}
    
    @self.settings_app.get("/status")
    def status():
        return {"bpm": bpm, "running": is_running}
    
    # Main loop
    while not stop_event.is_set():
        if is_running:
            # Do stuff
            pass
        time.sleep(0.01)

Lifecycle

When you call app.wrapped_run():

1. Web server starts (if custom_app_url is set) on a background thread
2. ReachyMini connects using context manager
3. Your run() method executes
4. On stop (Ctrl+C or stop_event.set()):
   • ReachyMini disconnects
   • Web server shuts down

def wrapped_run(self, *args, **kwargs) -> None:
    # Start settings server in background thread
    if self.settings_app is not None:
        # ... uvicorn server setup ...
        settings_app_t.start()
    
    try:
        with ReachyMini(*args, **kwargs) as reachy_mini:
            self.run(reachy_mini, self.stop_event)
    finally:
        if settings_app_t is not None:
            self.stop_event.set()
            settings_app_t.join()

Metronome App Structure

For our metronome, the structure will be:

from pydantic import BaseModel

class BpmUpdate(BaseModel):
    bpm: int

class TimeSignatureUpdate(BaseModel):
    beats: int

class ReachyMiniMetronome(ReachyMiniApp):
    custom_app_url = "http://0.0.0.0:8042"
    
    def run(self, reachy_mini: ReachyMini, stop_event: threading.Event):
        # State
        bpm = 120
        time_signature = 4
        current_beat = 1
        is_running = False
        
        # API endpoints
        @self.settings_app.post("/bpm")
        def set_bpm(data: BpmUpdate):
            nonlocal bpm
            bpm = max(40, min(208, data.bpm))
            return {"bpm": bpm}
        
        @self.settings_app.post("/time_signature")
        def set_time_signature(data: TimeSignatureUpdate):
            nonlocal time_signature, current_beat
            time_signature = data.beats
            current_beat = 1
            return {"time_signature": time_signature}
        
        @self.settings_app.post("/start")
        def start():
            nonlocal is_running
            is_running = True
            return {"running": True}
        
        @self.settings_app.post("/stop")
        def stop():
            nonlocal is_running
            is_running = False
            return {"running": False}
        
        @self.settings_app.get("/status")
        def get_status():
            return {
                "bpm": bpm,
                "time_signature": time_signature,
                "current_beat": current_beat,
                "running": is_running
            }
        
        # Main loop
        while not stop_event.is_set():
            # ... metronome logic ...
            time.sleep(0.01)