构建自定义 Providers

概览

本指南介绍如何为新的健康设备或平台构建自定义 Provider 集成。Provider 系统的设计目标是在保证安全与数据质量的前提下，让新增集成尽可能简单直观。

如果你只是想使用现有 Providers（如 Garmin、Whoop），请参见使用 Providers。

前置条件

在集成新的 Provider 之前，请确保你具备：

技术要求

Python 3.12+ 环境
可访问目标设备/服务的 API 文档
设备厂商提供的 OAuth 凭据（OAuth 1.0 或 OAuth 2.0）
熟悉 Python 的 async/await 模式
熟悉 REST APIs 与 JSON 数据格式

OAuth 凭据

从健康设备厂商获取 OAuth 凭据：

Client ID / Consumer Key
Client Secret / Consumer Secret
用于认证与数据访问的 API endpoints
访问健康数据所需的 OAuth scopes

数据库准备

你的 Provider 需要一个专用表来存储 raw 数据：

CREATE TABLE IF NOT EXISTS theta_ai.health_data_<provider> (
    id SERIAL PRIMARY KEY,
    create_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    update_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    is_del BOOLEAN DEFAULT FALSE,
    msg_id VARCHAR(255) UNIQUE NOT NULL,
    raw_data JSONB NOT NULL,
    theta_user_id VARCHAR(255) NOT NULL,
    external_user_id VARCHAR(255)
);

CREATE INDEX idx_health_data_<provider>_theta_user_id 
    ON theta_ai.health_data_<provider>(theta_user_id);
CREATE INDEX idx_health_data_<provider>_msg_id 
    ON theta_ai.health_data_<provider>(msg_id);

Provider Architecture

Directory Structure

connect/
├── __init__.py
└── theta/
    └── mirobody_<provider>/
        ├── __init__.py
        └── provider_<provider>.py

Class Hierarchy

BaseThetaProvider (from mirobody.pulse.theta.platform.base)
    ↓
ThetaYourProvider (your implementation)

完整实现细节请参见仓库中的 Provider Integration Guide。

Quick Start

创建 Provider 模块

为你的 Provider 创建一个新目录：

mkdir -p connect/theta/mirobody_yourprovider
touch connect/theta/mirobody_yourprovider/__init__.py
touch connect/theta/mirobody_yourprovider/provider_yourprovider.py

实现 Provider class

创建继承自 BaseThetaProvider 的 Provider class：

provider_yourprovider.py

from mirobody.pulse.theta.platform.base import BaseThetaProvider
from mirobody.pulse.theta.platform.indicator import StandardIndicator

class ThetaYourProvider(BaseThetaProvider):
    """Your health device provider"""
    
    @classmethod
    def factory(cls, **configs):
        """Factory method for provider instantiation"""
        return cls()
    
    @classmethod
    def info(cls) -> ProviderInfo:
        """Provider metadata"""
        return ProviderInfo(
            slug="theta_yourprovider",
            name="Your Provider",
            description="Integration with Your Health Device"
        )
    
    # Implement required methods...

配置 OAuth

在 config.yaml 中添加配置：

YOURPROVIDER_CLIENT_ID: 'your_client_id'
YOURPROVIDER_CLIENT_SECRET: 'your_client_secret'
YOURPROVIDER_REDIRECT_URL: 'http://localhost:18080/api/v1/pulse/theta/theta_yourprovider/callback'

测试 Provider

测试 OAuth flow 与数据拉取：

# Start the application
docker-compose up -d

# Test OAuth link
curl "http://localhost:18080/api/v1/pulse/theta/theta_yourprovider/link?user_id=test_user"

Implementation Checklist

Required Methods

你的 Provider 必须实现：

✅ factory() - Provider instantiation
✅ info() - Provider metadata
✅ link() - Initiate OAuth flow
✅ callback() - Handle OAuth callback
✅ unlink() - Disconnect provider
✅ pull_from_vendor_api() - Fetch data from API
✅ format_data() - Transform to standard format
✅ save_raw_data_to_db() - Persist raw data
✅ is_data_already_processed() - Check for duplicates

OAuth Implementation

根据 Provider 的要求选择 OAuth 版本：OAuth 1.0 (e.g., Garmin):

Request token
User authorization
Access token exchange

OAuth 2.0 (e.g., Whoop):

Authorization code
Token exchange
Token refresh handling

详细示例请参见 OAuth Implementation。

Data Mapping

将 vendor 特定数据映射到标准指标：

PROVIDER_INDICATOR_MAPPING = {
    "daily_steps": StandardIndicator.DAILY_STEPS,
    "sleep_time": StandardIndicator.DAILY_SLEEP_DURATION,
    "heart_rate": StandardIndicator.HEART_RATE,
}

转换模式请参见 Data Mapping。

Testing

充分测试你的 Provider：

Unit tests for each method
OAuth flow integration tests
Data transformation tests
Error handling tests

指南请参见 Provider Testing。

Key Concepts

Provider Lifecycle

Data Flow

Raw Data Fetch：从 vendor API 拉取数据
Raw Data Storage：保存到 provider 专用表
Transformation：转换为标准指标
Standard Storage：写入 th_series_data 表
Deduplication：跳过已处理记录

Configuration

在 config.yaml 中添加 Provider 配置：

config.yaml

# OAuth Settings
YOURPROVIDER_CLIENT_ID: 'your_client_id'
YOURPROVIDER_CLIENT_SECRET: 'your_client_secret'
YOURPROVIDER_REDIRECT_URL: 'http://localhost:18080/api/v1/pulse/theta/theta_yourprovider/callback'

# Optional: API Endpoints
YOURPROVIDER_AUTH_URL: 'https://auth.yourprovider.com'
YOURPROVIDER_TOKEN_URL: 'https://api.yourprovider.com/token'
YOURPROVIDER_API_BASE_URL: 'https://api.yourprovider.com/v1'

# Optional: OAuth Scopes
YOURPROVIDER_SCOPES: 'read:health read:profile'

# Optional: Performance
YOURPROVIDER_REQUEST_TIMEOUT: 30
YOURPROVIDER_CONCURRENT_REQUESTS: 5

Example: Minimal Provider

Here’s a minimal provider implementation:

from mirobody.pulse.theta.platform.base import BaseThetaProvider, ProviderInfo
from mirobody.pulse.theta.platform.indicator import StandardIndicator
from typing import Dict, List

class ThetaSimpleProvider(BaseThetaProvider):
    """Simple example provider"""
    
    @classmethod
    def factory(cls, **configs):
        return cls()
    
    @classmethod
    def info(cls) -> ProviderInfo:
        return ProviderInfo(
            slug="theta_simple",
            name="Simple Provider",
            description="Example provider integration"
        )
    
    async def link(self, user_id: str, return_url: str = None) -> Dict:
        """Initiate OAuth flow"""
        # Generate OAuth URL
        auth_url = self._build_auth_url(user_id)
        return {"link_web_url": auth_url}
    
    async def callback(self, user_id: str, **params) -> Dict:
        """Handle OAuth callback"""
        # Exchange code for tokens
        tokens = await self._exchange_token(params.get("code"))
        # Store tokens securely
        await self._store_tokens(user_id, tokens)
        return {"success": True}
    
    async def unlink(self, user_id: str) -> Dict:
        """Disconnect provider"""
        await self._delete_tokens(user_id)
        return {"success": True}
    
    async def pull_from_vendor_api(self, user_id: str) -> List[Dict]:
        """Fetch data from provider API"""
        tokens = await self._get_tokens(user_id)
        # Call vendor API
        data = await self._api_call("/user/health", tokens)
        return data
    
    def format_data(self, raw_data: Dict) -> List[StandardPulseData]:
        """Transform to standard format"""
        return [
            StandardPulseData(
                indicator=StandardIndicator.DAILY_STEPS,
                value=str(raw_data["steps"]),
                start_time=raw_data["date"],
                end_time=raw_data["date"]
            )
        ]

See Garmin Provider Example and Whoop Provider Example for complete implementations.

Detailed Documentation

For comprehensive implementation details, refer to these guides:

OAuth Implementation

Implement OAuth 1.0 or 2.0 flows

Data Mapping

Transform vendor data to standard format

Provider Testing

Test your provider integration

Full Integration Guide

Complete guide in repository

Standard Health Indicators

Map your provider’s data to these standard indicators:

Indicator	Description	Unit
`DAILY_STEPS`	Daily step count	steps
`DAILY_SLEEP_DURATION`	Total sleep time	milliseconds
`HEART_RATE`	Heart rate measurement	bpm
`DAILY_HEART_RATE_RESTING`	Resting heart rate	bpm
`HRV`	Heart rate variability	ms
`WEIGHT`	Body weight	grams
`DAILY_CALORIES_ACTIVE`	Active calories burned	kcal
`SLEEP_EFFICIENCY`	Sleep efficiency	percentage

See Standard Indicators for the complete list.

Contributing Your Provider

Once you’ve built a provider, consider contributing it back:

Test thoroughly

Ensure all OAuth flows work and data transforms correctly

Document your provider

Add configuration instructions and supported metrics

Submit pull request

Follow the Contributing Guide

See the complete Provider Integration Guide in the repository for detailed implementation instructions, including all required methods, OAuth patterns, data transformation examples, and best practices.

Next Steps

OAuth Implementation

Learn OAuth 1.0 and 2.0 flows

Data Mapping

Transform vendor data

Provider Testing

Test your integration

Examples

See real provider code

开发者指南

Providers 开发

部署

构建自定义 Providers

概览

前置条件

Provider Architecture

Directory Structure

Class Hierarchy

Quick Start

Implementation Checklist

Key Concepts

Provider Lifecycle

Data Flow

Configuration

Example: Minimal Provider

Detailed Documentation

OAuth Implementation

Data Mapping

Provider Testing

Full Integration Guide

Standard Health Indicators

Contributing Your Provider

Next Steps

OAuth Implementation

Data Mapping

Provider Testing

Examples

开发者指南

Providers 开发

部署

​概览

​前置条件

​Provider Architecture

​Directory Structure

​Class Hierarchy

​Quick Start

​Implementation Checklist

​Key Concepts

​Provider Lifecycle

​Data Flow

​Configuration

​Example: Minimal Provider

​Detailed Documentation

OAuth Implementation

Data Mapping

Provider Testing

Full Integration Guide

​Standard Health Indicators

​Contributing Your Provider

​Next Steps

OAuth Implementation

Data Mapping

Provider Testing

Examples

概览

前置条件

Provider Architecture

Directory Structure

Class Hierarchy

Quick Start

Implementation Checklist

Key Concepts

Provider Lifecycle

Data Flow

Configuration

Example: Minimal Provider

Detailed Documentation

Standard Health Indicators

Contributing Your Provider

Next Steps