VWED_server/CLAUDE.md

CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

Project Overview

VWED Task Module is a FastAPI-based Python system for managing and executing task workflows for AMR (Autonomous Mobile Robot) scheduling systems. The system provides a low-code configuration tool that allows users to design and configure complex robot task flows through a visual interface.

Development Commands

Application Startup

# Primary method - run the main application
python app.py

# Alternative method - using uvicorn directly
uvicorn app:app --host 0.0.0.0 --port 8000 --reload

Database Operations

# Run database migrations
python scripts/run_migration.py

# Generate new migrations
python scripts/generate_migration.py

# Initialize database
python scripts/init_db.py

Docker Deployment

# Build and run with Docker
docker build -t vwed-task:latest .
docker run -d -p 8000:8000 --name vwed-task-container vwed-task:latest

# Or use Docker Compose (recommended)
docker-compose up -d

Architecture Overview

The system follows a layered architecture with clear separation of concerns:

Core Layers

• API Layer (routes/): FastAPI REST endpoints for client communication
• Middleware Layer (middlewares/): Request logging, error handling, and cross-cutting concerns
• Business Logic Layer (services/): Core business logic including task execution and scheduling
• Data Layer (data/): ORM models, database session management, and data persistence
• Component Layer (components/): Extensible component system for task workflow building
• Configuration Layer (config/): System settings, database configuration, and error mappings

Key Services

• Task Execution Engine (services/execution/): Handles task lifecycle and execution
  • task_executor.py: Main task execution controller
  • block_executor.py: Individual task block execution
  • task_context.py: Execution context and variable management
  • handlers/: Specific handlers for different component types
• Enhanced Scheduler (services/enhanced_scheduler/): High-performance async task scheduling
• Intelligence Layer (services/intelligence/): AI-powered features (partial implementation)

Data Models

Key models in data/models/:

• taskdef.py: Task definition and configuration
• taskrecord.py: Task execution records
• blockrecord.py: Individual block execution records
• tasktemplate.py: Reusable task templates
• calldevice.py: Device integration configurations

Key Features

Task Management

• Visual task flow designer with drag-and-drop interface
• Version control for task configurations
• Task templates for reusability
• Real-time execution monitoring

Component System

Components are registered in config/components/ and include:

• Foundation components (basic operations)
• Robot scheduling components
• HTTP request components
• Storage location management
• Progress tracking components
• Script execution components

Device Integration

• Modbus protocol support (modbusconfig.py)
• Generic device calling interface (calldevice.py)
• Extensible communication protocols

Development Guidelines

Database Schema Changes

1. Modify models in data/models/
2. Generate migration: python scripts/generate_migration.py
3. Apply migration: python scripts/run_migration.py

Adding New Components

1. Define component configuration in config/components/
2. Implement handler in services/execution/handlers/
3. Register component in the component system

Task Execution Flow

1. Task definition stored in taskdef table
2. Execution creates taskrecord entry
3. Individual blocks create blockrecord entries
4. Context variables managed through task_context.py
5. Component handlers execute specific business logic

Configuration Management

• Environment-specific settings in config/settings.py
• Database configuration in config/database_config.py
• Error messages centralized in config/error_messages.py

Important Technical Details

Async Task Scheduling

The system uses a custom enhanced scheduler (services/enhanced_scheduler/) that:

• Maintains worker pools for concurrent task execution
• Provides priority-based task queuing
• Handles task persistence and recovery
• Manages worker lifecycle

Component Architecture

Components follow a registry pattern:

• Each component type has a handler class
• Handlers implement standard execution interface
• Components are configurable through JSON definitions
• Extensible for new component types

Database Session Management

• Uses SQLAlchemy ORM with session management in data/session.py
• Supports both sync and async database operations
• Automatic connection pooling and cleanup

Testing and Debugging

Test Files

• Basic tests in tests/ directory
• Test data and fixtures available
• Integration tests for API endpoints

Logging

• Centralized logging configuration in utils/logger.py
• Application logs stored in logs/ directory
• Structured logging for debugging task execution

API Documentation

• Swagger UI available at http://localhost:8000/docs
• Comprehensive API documentation in VWED任务模块接口文档/

Common Troubleshooting

Database Issues

• Check database connection in config/database_config.py
• Verify database migrations are up to date
• Review logs in logs/app.log

Task Execution Problems

• Monitor task status through API endpoints
• Check execution logs for specific error messages
• Verify component configurations are correct

Performance Optimization

• Adjust scheduler worker counts in settings
• Monitor database connection pool usage
• Review task complexity and component efficiency