Project files

HomeBot New Application Development Guide

This document guides developers on how to develop new application modules based on the HomeBot framework.

Table of Contents

1. Quick Start
2. Application Structure
3. Core Components
4. Development Steps
5. Complete Example
6. Debugging and Testing

Quick Start

Prerequisites

# 1. Ensure dependencies are installed
cd homebot
pip install -e .

# 2. Start the basic services
cd software/src

# Terminal 1: Start the chassis service
python -m services.motion_service.chassis_service

# Terminal 2: Start the vision service (if needed)
python -m services.vision_service

Creating a Minimal Application (3-Minute Quick Start)

cd software/src/applications

# Create a new application directory
mkdir my_app
cd my_app

# Create required files
touch __init__.py __main__.py app.py

app.py – Core application logic:

"""My First HomeBot Application"""
import time
from common.logging import get_logger
from services.motion_service.chassis_arbiter import ChassisArbiterClient

logger = get_logger(__name__)

class MyFirstApp:
    """Example application: make the robot draw a square

    Note: The chassis controller has a 1000 ms timeout; commands must be sent continuously to maintain motion.
    """

    def __init__(self):
        # Initialize chassis client
        self.chassis = ChassisArbiterClient("tcp://localhost:5556")
        self.running = False
        # Control frequency (Hz); must be > 1 Hz to avoid timeout
        self.control_rate = 20  # 20 Hz = send every 50 ms

    def move_forward(self, duration=2.0, speed=0.3):
        """Move forward for the specified duration.

        Args:
            duration: Movement time in seconds.
            speed: Forward speed in m/s.
        """
        logger.info(f"Moving forward for {duration} s")
        start_time = time.time()

        # Continuously send commands to retain control (heartbeat mode)
        while time.time() - start_time < duration:
            self.chassis.send_command(
                vx=speed, vy=0.0, vz=0.0,
                source="auto", priority=3
            )
            time.sleep(1.0 / self.control_rate)

        self.stop()

    def turn_right(self, duration=1.0, speed=1.57):
        """Turn right for the specified duration.

        Args:
            duration: Turn time in seconds.
            speed: Angular velocity in rad/s; 1.57 ≈ 90°/s.
        """
        logger.info(f"Turning right for {duration} s")
        start_time = time.time()

        # Continuously send commands to retain control
        while time.time() - start_time < duration:
            self.chassis.send_command(
                vx=0.0, vy=0.0, vz=-speed,  # Negative = right turn
                source="auto", priority=3
            )
            time.sleep(1.0 / self.control_rate)

        self.stop()

    def stop(self):
        """Stop the robot."""
        self.chassis.send_command(
            vx=0.0, vy=0.0, vz=0.0,
            source="auto", priority=3
        )

    def run(self):
        """Main loop: draw a square."""
        logger.info("Starting square path execution")
        self.running = True

        try:
            for i in range(4):
                if not self.running:
                    break
                logger.info(f"Side {i+1}")
                self.move_forward(2.0)
                time.sleep(0.5)
                self.turn_right(1.0)
                time.sleep(0.5)

            logger.info("Task complete")

        except KeyboardInterrupt:
            logger.info("User interrupted")
        finally:
            self.stop()
            self.running = False

⚠️ Important: The chassis service has a 1000 ms (1 s) timeout. After the last command is sent, if no new command arrives within 1 s, control is automatically released and the chassis will stop. Therefore, movement operations must use continuous sending (heartbeat) mode, not a simple “send → delay → stop” pattern.

__main__.py – Application entry point:

#!/usr/bin/env python3
"""Application startup entry point"""
from .app import MyFirstApp

if __name__ == "__main__":
    app = MyFirstApp()
    app.run()

__init__.py – Package marker (can be empty):

"""My First App package"""

Running the Application

cd software/src
python -m applications.my_app

Application Structure

Standard Directory Structure

applications/my_app/
├── __init__.py          # Package initialization (optional exports)
├── __main__.py          # Application entry point
├── app.py               # Main application logic
├── detector.py          # Detector (if needed)
├── controller.py        # Controller (if needed)
├── utils.py             # Utility functions
└── README.md            # Application description

Naming Conventions

Core Components

1. Chassis Control (Required)

from services.motion_service.chassis_arbiter import ChassisArbiterClient
import time

# Initialize the client
chassis = ChassisArbiterClient("tcp://localhost:5556")

# ⚠️ Important: The chassis has a 1000ms timeout mechanism and requires continuous command transmission (heartbeat mode).
# Incorrect method (will time out and stop):
# chassis.send_command(vx=0.5, vy=0, vz=0, source="auto", priority=3)
# time.sleep(2)  # Exceeds 1 second, chassis has already stopped automatically!

# Correct method (continuous transmission):
start_time = time.time()
while time.time() - start_time < 2.0:  # Continuous for 2 seconds
    chassis.send_command(
        vx=0.5,      # Linear velocity X (m/s), positive for forward
        vy=0.0,      # Linear velocity Y (m/s), positive for left
        vz=0.3,      # Angular velocity Z (rad/s), positive for counterclockwise
        source="auto",   # Control source: web/voice/auto/emergency
        priority=3       # Priority: 1=web, 2=voice, 3=auto, 4=emergency
    )
    time.sleep(0.05)  # 20Hz, ensures it does not exceed the 1-second timeout

# Send stop command
chassis.send_command(vx=0, vy=0, vz=0, source="auto", priority=3)

Chassis Timeout Mechanism Explanation:

• Timeout duration: 1000ms (1 second)
• Mechanism: If no new command is received within 1 second after the last command → control is automatically released → chassis stops
• Solution: Commands must be sent continuously at a frequency >1Hz (recommended 20-50Hz)

Control Source Priority:

emergency(4) > auto(3) > voice(2) > web(1)

2. Vision Subscription (Optional)

from services.vision_service import VisionSubscriber
import cv2

# Initialize subscription
vision = VisionSubscriber("tcp://localhost:5560")
vision.start()

# Read image frame
frame_id, frame = vision.read_frame()
if frame is not None:
    # frame is a numpy.ndarray (BGR format)
    cv2.imshow("Camera", frame)

# Stop subscription
vision.stop()

3. Configuration Management

from configs import get_config

# Get global configuration
config = get_config()

# Access configurations for each module
print(config.chassis.serial_port)      # Chassis serial port
print(config.camera.width)             # Camera width
print(config.human_follow.model_path)  # Model path

# Custom application configuration (recommended)
from dataclasses import dataclass

@dataclass
class MyAppConfig:
    loop_count: int = 5
    speed: float = 0.3
    enable_debug: bool = False

4. Logging

from common.logging import get_logger

# Create a logger
logger = get_logger(__name__)

# Logs at different levels
logger.debug("Debug message")
logger.info("General information")
logger.warning("Warning message")
logger.error("Error message")
logger.critical("Critical error")

Development Steps

Step 1: Planning the Application Architecture

Determine the components required for the application:

• Is visual input needed?
• Is chassis control required?
• Is voice interaction needed?
• Is robotic arm control required?
• State machine design

Step 2: Create the Application Framework

mkdir -p software/src/applications/my_app
touch software/src/applications/my_app/__init__.py
touch software/src/applications/my_app/__main__.py

Step 3: Implement the Main Class

Reference template:

"""Application template"""
import threading
import time
from typing import Optional
from enum import Enum

from common.logging import get_logger
from configs import get_config
from services.motion_service.chassis_arbiter import ChassisArbiterClient
from services.vision_service import VisionSubscriber

logger = get_logger(__name__)

class AppState(Enum):
    """Application state"""
    IDLE = "idle"
    RUNNING = "running"
    PAUSED = "paused"
    ERROR = "error"

class MyApplication:
    """
    New application template

    Function description:
    1. xxx
    2. xxx

    Usage:
        cd software/src
        python -m applications.my_app
    """

    def __init__(self, config=None):
        """Initialize the application"""
        self.config = config or get_config()
        self.state = AppState.IDLE
        self.running = False

        # Components (lazy initialization)
        self.chassis: Optional[ChassisArbiterClient] = None
        self.vision: Optional[VisionSubscriber] = None

        # Thread control
        self._stop_event = threading.Event()

        logger.info("Application initialization complete")

    def initialize(self) -> bool:
        """
        Initialize all components

        Returns:
            bool: Whether initialization was successful
        """
        try:
            logger.info("Initializing...")

            # Initialize chassis client
            self.chassis = ChassisArbiterClient("tcp://localhost:5556")
            logger.info("✓ Chassis client connected")

            # Uncomment if vision is needed
            # self.vision = VisionSubscriber("tcp://localhost:5560")
            # self.vision.start()
            # logger.info("✓ Vision subscription started")

            logger.info("Initialization complete")
            return True

        except Exception as e:
            logger.error(f"Initialization failed: {e}")
            self.state = AppState.ERROR
            return False

    def process(self):
        """
        Main processing logic (called every frame)
        Subclasses should override this method

        Note: If chassis control is involved, continuous commands must be sent in this method
        to avoid the 1000ms timeout mechanism that would cause automatic stopping
        """
        pass

    def run(self):
        """Main loop"""
        if not self.initialize():
            return

        self.running = True
        self.state = AppState.RUNNING
        self._stop_event.clear()

        logger.info("=" * 50)
        logger.info("Application started, press Ctrl+C to stop")
        logger.info("=" * 50)

        # Control frequency (Hz), must be > 1Hz to avoid chassis timeout
        control_rate = 20
        dt = 1.0 / control_rate

        try:
            while self.running and not self._stop_event.is_set():
                loop_start = time.time()

                self.process()

                # Precisely control loop frequency
                elapsed = time.time() - loop_start
                sleep_time = dt - elapsed
                if sleep_time > 0:
                    time.sleep(sleep_time)

        except KeyboardInterrupt:
            logger.info("User interrupted")
        except Exception as e:
            logger.error(f"Runtime exception: {e}")
            self.state = AppState.ERROR
        finally:
            self.stop()

    def pause(self):
        """Pause the application"""
        if self.state == AppState.RUNNING:
            self.state = AppState.PAUSED
            self.stop_chassis()
            logger.info("Application paused")

    def resume(self):
        """Resume the application"""
        if self.state == AppState.PAUSED:
            self.state = AppState.RUNNING
            logger.info("Application resumed")

    def stop(self):
        """Stop the application"""
        logger.info("Stopping application...")
        self.running = False
        self._stop_event.set()
        self.state = AppState.IDLE

        # Stop chassis
        self.stop_chassis()

        # Release resources
        if self.vision:
            self.vision.stop()

        logger.info("Application stopped")

    def stop_chassis(self):
        """Stop chassis movement"""
        if self.chassis:
            try:
                self.chassis.send_command(
                    vx=0, vy=0, vz=0,
                    source="auto", priority=3
                )
            except Exception as e:
                logger.warning(f"Failed to stop chassis: {e}")

def main():
    """Entry function"""
    import argparse

    parser = argparse.ArgumentParser(description='My Application')
    parser.add_argument('--debug', '-d', action='store_true',
                       help='Enable debug mode')

    args = parser.parse_args()

    app = MyApplication()
    app.run()

if __name__ == "__main__":
    main()

Step 4: Adding Configuration (Optional)

Edit software/src/configs/config.py:

@dataclass
class MyAppConfig:
    """Configuration for my application"""
    param1: float = 0.5
    param2: int = 100
    enabled: bool = True

@dataclass
class Config:
    """Global configuration"""
    # ... existing configuration ...
    my_app: MyAppConfig = field(default_factory=MyAppConfig)

Step 5: Testing and Debugging

# Run the application
cd software/src
python -m applications.my_app

# With debug output
python -m applications.my_app --debug

Complete Example

Example: Automatic Obstacle Avoidance Application

"""Automatic obstacle avoidance application - using distance estimation"""
import time
import numpy as np
from typing import Optional, List
from dataclasses import dataclass

from common.logging import get_logger
from services.motion_service.chassis_arbiter import ChassisArbiterClient
from services.vision_service import VisionSubscriber

logger = get_logger(__name__)

@dataclass
class Obstacle:
    """Obstacle information"""
    x: int      # Image coordinate x
    y: int      # Image coordinate y
    width: int  # Width
    distance: float  # Estimated distance (m)

class AvoidanceApp:
    """
    Simple obstacle avoidance application

    Features:
    - Move forward
    - Turn when obstacles are detected
    - Safe distance: 0.5 meters

    Note: Use a continuous control loop (20Hz) to prevent chassis timeout shutdown
    """

    def __init__(self):
        self.chassis: Optional[ChassisArbiterClient] = None
        self.vision: Optional[VisionSubscriber] = None

        # Parameters
        self.safe_distance = 0.5   # Safe distance (meters)
        self.forward_speed = 0.3   # Forward speed
        self.turn_speed = 0.8      # Turning speed
        self.control_rate = 20     # Control frequency (Hz), must be > 1Hz

        self.running = False
        # Current speed (for smoothing)
        self.current_vx = 0.0
        self.current_vz = 0.0

    def initialize(self) -> bool:
        """Initialize"""
        try:
            self.chassis = ChassisArbiterClient("tcp://localhost:5556")
            self.vision = VisionSubscriber("tcp://localhost:5560")
            self.vision.start()
            logger.info("Obstacle avoidance application initialized")
            return True
        except Exception as e:
            logger.error(f"Initialization failed: {e}")
            return False

    def detect_obstacles(self, frame) -> List[Obstacle]:
        """
        Detect obstacles (simplified example)
        In actual applications, use depth cameras or stereo vision
        """
        obstacles = []
        h, w = frame.shape[:2]

        # Example: Detect large area at the bottom of the frame
        # In practice, use semantic segmentation or depth estimation
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

        # Simple ground detection (example)
        bottom_region = gray[int(h*0.7):, :]

        # If there is an object in the lower region (large pixel value variation)
        if np.std(bottom_region) > 50:
            # Estimate distance (based on area)
            distance = 1.0  # Simplified estimation
            obstacles.append(Obstacle(
                x=w//2, y=int(h*0.8),
                width=w//3, distance=distance
            ))

        return obstacles

    def decide_action(self, obstacles: List[Obstacle]) -> tuple:
        """
        Decision making

        Returns:
            (vx, vy, vz): Target velocity command
        """
        if not obstacles:
            # No obstacles, move forward
            return self.forward_speed, 0.0, 0.0

        # Find the closest obstacle
        closest = min(obstacles, key=lambda o: o.distance)

        if closest.distance > self.safe_distance:
            # Distance is safe, continue forward
            return self.forward_speed, 0.0, 0.0

        # Need to avoid obstacle
        logger.info(f"Obstacle detected, distance: {closest.distance:.2f}m")

        # Decide turning direction based on obstacle position
        frame_center = 640  # Assume resolution
        if closest.x < frame_center:
            # Obstacle on the left, turn right
            return 0.0, 0.0, -self.turn_speed
        else:
            # Obstacle on the right, turn left
            return 0.0, 0.0, self.turn_speed

    def smooth_velocity(self, target_vx: float, target_vz: float, alpha: float = 0.3) -> tuple:
        """Velocity smoothing to avoid sudden changes"""
        self.current_vx = alpha * target_vx + (1 - alpha) * self.current_vx
        self.current_vz = alpha * target_vz + (1 - alpha) * self.current_vz
        return self.current_vx, self.current_vz

    def run(self):
        """Main loop - continuous control mode (to avoid timeout)"""
        if not self.initialize():
            return

        self.running = True
        logger.info("Obstacle avoidance application started")

        # Control period (seconds)
        dt = 1.0 / self.control_rate

        try:
            while self.running:
                loop_start = time.time()

                # Read image
                frame_id, frame = self.vision.read_frame()
                if frame is None:
                    time.sleep(dt)
                    continue

                # Detect obstacles
                obstacles = self.detect_obstacles(frame)

                # Decision (target velocity)
                target_vx, _, target_vz = self.decide_action(obstacles)

                # Velocity smoothing
                vx, vz = self.smooth_velocity(target_vx, target_vz)

                # ⚠️ Key: Continuously send commands to maintain control (heartbeat mode)
                self.chassis.send_command(
                    vx=vx, vy=0.0, vz=vz,
                    source="auto", priority=3
                )

                # Precisely control loop frequency
                elapsed = time.time() - loop_start
                sleep_time = dt - elapsed
                if sleep_time > 0:
                    time.sleep(sleep_time)

        except KeyboardInterrupt:
            logger.info("User interrupted")
        finally:
            self.stop()

    def stop(self):
        """Stop the application"""
        logger.info("Stopping obstacle avoidance application...")
        self.running = False

        # Send stop command
        if self.chassis:
            self.chassis.send_command(
                vx=0, vy=0, vz=0,
                source="auto", priority=3
            )

    if self.vision:
        self.vision.stop()

    logger.info("Obstacle avoidance application stopped")

def main(): app = AvoidanceApp() app.run()

if name == "main": import cv2 # delayed import main()

Debugging and Testing

1. Log Debugging

# Set log level
import logging
logging.getLogger().setLevel(logging.DEBUG)

# Add logs at key locations
logger.debug(f"Current speed: vx={vx}, vz={vz}")
logger.debug(f"Detected targets: {len(detections)}")

2. Simulation Mode

If the hardware is unavailable, you can create a mock client:

class MockChassisClient:
    """Mock chassis client for testing"""

    def send_command(self, vx, vy, vz, source, priority):
        print(f"[Mock] Speed: vx={vx}, vy={vy}, vz={vz}")
        return type('Response', (), {'success': True})()

3. Unit Testing

Create tests under software/src/tests/:

# tests/test_my_app.py
import unittest
from applications.my_app.app import MyApplication

class TestMyApp(unittest.TestCase):

    def test_initialization(self):
        app = MyApplication()
        self.assertEqual(app.state.value, "idle")

    def test_state_transition(self):
        app = MyApplication()
        # Test state transition
        app.state = AppState.RUNNING
        self.assertEqual(app.state.value, "running")

if __name__ == "__main__":
    unittest.main()

Run the tests:

cd software/src
python -m tests.test_my_app

4. Performance Analysis

import time

# Measure processing time
start = time.time()
result = process_frame(frame)
elapsed = (time.time() - start) * 1000
logger.debug(f"Processing time: {elapsed:.1f}ms")

Important Notes on Chassis Control

Timeout Mechanism

The HomeBot chassis service implements a control authority timeout protection mechanism:

❌ Incorrect Example (Will Time Out and Stop)

# Incorrect: Sending a command and then waiting
chassis.send_command(vx=0.5, vy=0, vz=0, source="auto", priority=3)
time.sleep(2)  # Exceeds 1 second; the chassis has already stopped automatically!
chassis.send_command(vx=0, vy=0, vz=0, source="auto", priority=3)  # Stop command is now ineffective

✅ Correct Example (Heartbeat Mode)

# Correct: Continuously send commands to maintain movement
import time

start_time = time.time()
duration = 2.0  # Run for 2 seconds
control_rate = 20  # 20Hz

while time.time() - start_time < duration:
    chassis.send_command(vx=0.5, vy=0, vz=0, source="auto", priority=3)
    time.sleep(1.0 / control_rate)  # Send every 50ms

# Finally, send a stop command
chassis.send_command(vx=0, vy=0, vz=0, source="auto", priority=3)

Control Mode Comparison

Best Practices

1. Priority Usage Specifications

   • web (1): Web remote control
   • voice (2): Voice control
   • auto (3): Automatic mode (following, obstacle avoidance, etc.)
   • emergency (4): Emergency stop

2. Chassis Control Specifications

   • Never use the "send → delay → stop" mode
   • Always use continuous sending (heartbeat) mode, frequency > 1Hz
   • Explicitly send zero-speed commands when stopping
   • Use speed smoothing to avoid sudden changes

3. Resource Management

   • Use try/finally or context managers to ensure resource release
   • Ensure the chassis stops in case of exceptions

4. Error Handling

   • Reconnect when network anomalies occur
   • Gracefully degrade when services are unavailable
   • Stop and retry when the chassis is unresponsive

5. State Machine Design

   • Clearly define state transitions
   • Avoid state confusion
   • Stop sending commands when in the stop state

6. Code Organization

   • Single Responsibility Principle
   • Separate detection/tracking/control
   • Centralize control logic in a single loop

Frequently Asked Questions

Q: Why doesn’t the chassis respond to commands?

A: Check the following:

1. Whether the chassis service has started
2. Whether the control source priority is set correctly
3. Whether a higher-priority control source is occupying the chassis

Q: The robot moves for a moment and then stops?

A: This is caused by the 1000 ms timeout mechanism. The chassis service will automatically stop 1 second after the last command. Solutions:

• Use the heartbeat mode (continuous sending), with a control frequency > 1 Hz
• Refer to the “Important Notes on Chassis Control” section in this document
• Incorrect example: send_command() → sleep(2) → stop()
• Correct example: Continuously call send_command() at 20 Hz inside a while loop

Q: How do I obtain camera images?

A: Subscribe to the image stream published by the vision service using VisionSubscriber.

Q: How can the application communicate with the web frontend?

A: You can:

1. Communicate indirectly via the chassis service
2. Add a custom ZeroMQ communication channel
3. Use shared state or a database

Q: How do I debug vision algorithms?

A: Run the program with the --display parameter to open a debug window that visualizes detection results.

Document version: 1.0 Last updated: 2026-03-13